#!/usr/bin/env perl

# test_mergeJoinListings.pl
#
# Tests validateBackup's listing merge-join core: readListingRecord, mergeJoinListings,
# inMtimeWindow and classifyPair.
#
# validateBackup is a script, not a requireable module (unguarded top-level logic runs on load),
# so the functions under test are copied here verbatim, per this project's "extract a sub into a
# stub harness" testing convention (see test_destroyedSourceDataset.pl). recordFinding and
# countActiveFile are replaced with local stubs that collect into arrays instead of writing a
# findings file. No live ZFS, ssh or sort is required - the fixtures are plain text files.
#
# IF YOU CHANGE validateBackup's copies of these functions, change them here too.
#
# Author: R. W. Rodolico <rodo@dailydata.net>
# Created: September 2026

use strict;
use warnings;
use FindBin;
use lib "$FindBin::Bin/..";
use File::Temp qw(tempdir);

my ( $passed, $failed ) = ( 0, 0 );

sub ok {
   my ( $condition, $description ) = @_;
   if ($condition) { print "  PASS: $description\n"; $passed++; }
   else            { print "  FAIL: $description\n"; $failed++; }
}

my $tempDir = tempdir( CLEANUP => 1 );

## Write a listing fixture and return its path. Lines are given already in the order they should
## appear in the file, so a test can deliberately supply a WRONG order.
sub writeListing {
   my ( $name, @lines ) = @_;
   my $path = "$tempDir/$name";
   open( my $fh, '>', $path ) or die "cannot write $path: $!";
   binmode $fh;
   print { $fh } "$_\n" foreach @lines;
   close($fh);
   return $path;
}

## Build a record line in the real format: path first, then size, then mtime, tab separated.
sub rec {
   my ( $path, $size, $mtime ) = @_;
   return join( "\t", $path, $size, $mtime );
}

# ---------------------------------------------------------------------------------------------
# Functions under test, copied verbatim from validateBackup
# ---------------------------------------------------------------------------------------------

sub readListingRecord {
   my ($fh) = @_;
   my $line = <$fh>;
   return undef unless defined $line;
   chomp $line;

   my ( $path, $size, $mtime ) = split( /\t/, $line, 3 );
   my $malformed =
      ( !defined $mtime || $size !~ /^\d+$/ || $mtime !~ /^\d+$/ ) ? 1 : 0;
   return {
      path      => defined $path ? $path : '',
      size      => $malformed ? 0 : $size + 0,
      mtime     => $malformed ? 0 : $mtime + 0,
      malformed => $malformed,
   };
}

sub mergeJoinListings {
   my ( $activeFile, $backupFile, $handler ) = @_;

   open( my $activeFh, '<', $activeFile ) or return ( 0, "cannot read '$activeFile': $!" );
   open( my $backupFh, '<', $backupFile )
      or do { close($activeFh); return ( 0, "cannot read '$backupFile': $!" ) };
   binmode $activeFh;
   binmode $backupFh;

   my $activeRecord = readListingRecord($activeFh);
   my $backupRecord = readListingRecord($backupFh);
   while ( defined $activeRecord || defined $backupRecord ) {
      if ( !defined $backupRecord ) {
         $handler->( 'activeOnly', $activeRecord, undef );
         $activeRecord = readListingRecord($activeFh);
      }
      elsif ( !defined $activeRecord ) {
         $handler->( 'backupOnly', undef, $backupRecord );
         $backupRecord = readListingRecord($backupFh);
      }
      elsif ( $activeRecord->{path} lt $backupRecord->{path} ) {
         $handler->( 'activeOnly', $activeRecord, undef );
         $activeRecord = readListingRecord($activeFh);
      }
      elsif ( $activeRecord->{path} gt $backupRecord->{path} ) {
         $handler->( 'backupOnly', undef, $backupRecord );
         $backupRecord = readListingRecord($backupFh);
      }
      else {
         $handler->( 'both', $activeRecord, $backupRecord );
         $activeRecord = readListingRecord($activeFh);
         $backupRecord = readListingRecord($backupFh);
      }
   }

   close($activeFh);
   close($backupFh);
   return ( 1, '' );
}

sub inMtimeWindow {
   my ( $window, $mtime ) = @_;
   return 1 unless $window->{active};
   return 1 unless defined $mtime;
   return 0 if defined $window->{minMtime} && $mtime < $window->{minMtime};
   return 0 if defined $window->{maxMtime} && $mtime > $window->{maxMtime};
   return 1;
}

sub newCounters {
   return {
      fileCount        => 0,
      totalFileSize    => 0,
      filesCompared    => 0,
      filesMatched     => 0,
      backupOnlyCount  => 0,
      filesSampled     => 0,
      filesValidated   => 0,
      listingWarnings  => 0,
      listingMalformed => 0,
      errorCounts      => {
         missing             => 0,
         extra               => 0,
         sizeDiff            => 0,
         mtimeDiff           => 0,
         checksumDiff        => 0,
         checksumUnavailable => 0,
      },
   };
}

sub countActiveFile {
   my ( $counters, $record ) = @_;
   $counters->{fileCount}++;
   $counters->{totalFileSize} += $record->{size};
   return;
}

sub classifyPair {
   my ( $config, $runState, $pair, $kind, $activeRec, $backupRec ) = @_;
   my $counters = $pair->{counters};

   $counters->{listingMalformed}++
      if ( $activeRec && $activeRec->{malformed} ) || ( $backupRec && $backupRec->{malformed} );

   if ( $kind eq 'activeOnly' ) {
      return unless inMtimeWindow( $runState->{mtimeWindow}, $activeRec->{mtime} );
      countActiveFile( $counters, $activeRec );
      recordFinding( $config, $runState, $pair, 'missing', $activeRec->{path}, $activeRec, undef, '' );
      return;
   }

   if ( $kind eq 'backupOnly' ) {
      return unless inMtimeWindow( $runState->{mtimeWindow}, $backupRec->{mtime} );
      $counters->{backupOnlyCount}++;
      recordFinding( $config, $runState, $pair, 'extra', $backupRec->{path}, undef, $backupRec, '' );
      return;
   }

   return unless inMtimeWindow( $runState->{mtimeWindow}, $activeRec->{mtime} );
   countActiveFile( $counters, $activeRec );
   $counters->{filesCompared}++;

   my $mismatch = 0;
   if ( $activeRec->{size} != $backupRec->{size} ) {
      recordFinding( $config, $runState, $pair, 'sizeDiff', $activeRec->{path}, $activeRec, $backupRec,
         "active $activeRec->{size} vs backup $backupRec->{size}" );
      $mismatch = 1;
   }
   if ( abs( $activeRec->{mtime} - $backupRec->{mtime} ) > $config->{mtimeSlack} ) {
      recordFinding( $config, $runState, $pair, 'mtimeDiff', $activeRec->{path}, $activeRec, $backupRec,
         "active $activeRec->{mtime} vs backup $backupRec->{mtime}" );
      $mismatch = 1;
   }
   $counters->{filesMatched}++ unless $mismatch;
   return;
}

# ---------------------------------------------------------------------------------------------
# Stub: collects findings in memory instead of writing the findings file
# ---------------------------------------------------------------------------------------------

my @recordedFindings;

sub recordFinding {
   my ( $config, $runState, $pair, $type, $path, $activeRec, $backupRec, $detail ) = @_;
   $pair->{counters}->{errorCounts}->{$type}++;
   push @recordedFindings, { type => $type, path => $path, detail => $detail };
   return;
}

## Run a join over two listings and return the collected ( kinds, counters, findings ).
sub runJoin {
   my ( $activeLines, $backupLines, %options ) = @_;
   @recordedFindings = ();
   my $config = { mtimeSlack => $options{mtimeSlack} // 0 };
   my $runState = {
      mtimeWindow => $options{window} // { active => 0, minMtime => undef, maxMtime => undef },
   };
   my $pair = { counters => newCounters(), activeDataset => 'a', backupDataset => 'b' };
   my @kinds;

   my $activeFile = writeListing( "active.$options{label}", @$activeLines );
   my $backupFile = writeListing( "backup.$options{label}", @$backupLines );
   my ( $joined, $message ) = mergeJoinListings(
      $activeFile, $backupFile,
      sub {
         my ( $kind, $a, $b ) = @_;
         push @kinds, $kind;
         classifyPair( $config, $runState, $pair, $kind, $a, $b );
      }
   );
   return {
      joined   => $joined,
      message  => $message,
      kinds    => \@kinds,
      counters => $pair->{counters},
      findings => [@recordedFindings],
   };
}

# ---------------------------------------------------------------------------------------------
print "\n=== readListingRecord ===\n";
# ---------------------------------------------------------------------------------------------
{
   my $path = writeListing( 'records',
      rec( './a.txt',       100, 1757500000 ),
      rec( './with space',  200, 1757500001 ),
      "./missing-fields",
      rec( './bad-size', 'x',  1757500002 ),
      rec( './bad-mtime', 300, 'y' ),
   );
   open( my $fh, '<', $path ) or die;
   binmode $fh;

   my $r = readListingRecord($fh);
   ok( $r->{path} eq './a.txt' && $r->{size} == 100 && $r->{mtime} == 1757500000 && !$r->{malformed},
      'well-formed record parses path, size and mtime' );

   $r = readListingRecord($fh);
   ok( $r->{path} eq './with space' && $r->{size} == 200 && !$r->{malformed},
      'path containing a space parses correctly (tab is the delimiter, not whitespace)' );

   $r = readListingRecord($fh);
   ok( $r->{malformed} == 1, 'line with no tab-separated fields is flagged malformed' );

   $r = readListingRecord($fh);
   ok( $r->{malformed} == 1, 'non-integer size is flagged malformed' );

   $r = readListingRecord($fh);
   ok( $r->{malformed} == 1, 'non-integer mtime is flagged malformed' );

   ok( !defined readListingRecord($fh), 'returns undef at EOF' );
   close($fh);
}

# ---------------------------------------------------------------------------------------------
print "\n=== the sort-order assumption the join depends on ===\n";
# ---------------------------------------------------------------------------------------------
{
   # The whole pipeline rests on this: a byte-wise (LC_ALL=C) sort places a file whose name
   # extends a directory's name BEFORE that directory's contents, because '.' (0x2E) sorts below
   # '/' (0x2F). BSD 'find -s' produces the OPPOSITE order, which is why the pipeline must sort
   # and must never feed find -s output straight into the join.
   ok( './Documentation.md' lt './Documentation/index.html',
      "byte order places './Documentation.md' before './Documentation/index.html' ('.' < '/')" );

   my @sorted = sort( './Documentation/index.html', './Documentation.md' );
   ok( $sorted[0] eq './Documentation.md',
      "Perl's default sort agrees with that ordering (no locale in effect)" );
}

# ---------------------------------------------------------------------------------------------
print "\n=== mergeJoinListings: basic pairing ===\n";
# ---------------------------------------------------------------------------------------------
{
   my $result = runJoin(
      [ rec( './a', 10, 100 ), rec( './b', 20, 200 ) ],
      [ rec( './a', 10, 100 ), rec( './b', 20, 200 ) ],
      label => 'identical'
   );
   ok( $result->{joined}, 'join succeeds on identical listings' );
   ok( join( ',', @{ $result->{kinds} } ) eq 'both,both', 'identical listings yield only "both"' );
   ok( $result->{counters}->{filesMatched} == 2, 'both files counted as matched' );
   ok( $result->{counters}->{fileCount} == 2, 'fileCount counts active-side files' );
   ok( $result->{counters}->{totalFileSize} == 30, 'totalFileSize sums active-side sizes' );
   ok( scalar @{ $result->{findings} } == 0, 'no findings for identical listings' );
}
{
   my $result = runJoin(
      [ rec( './a', 10, 100 ), rec( './only-active', 10, 100 ) ],
      [ rec( './a', 10, 100 ) ],
      label => 'activeonly'
   );
   ok( join( ',', @{ $result->{kinds} } ) eq 'both,activeOnly', 'trailing active-only file detected' );
   ok( $result->{counters}->{errorCounts}->{missing} == 1, 'active-only file recorded as missing' );
}
{
   my $result = runJoin(
      [ rec( './a', 10, 100 ) ],
      [ rec( './a', 10, 100 ), rec( './only-backup', 10, 100 ) ],
      label => 'backuponly'
   );
   ok( join( ',', @{ $result->{kinds} } ) eq 'both,backupOnly', 'trailing backup-only file detected' );
   ok( $result->{counters}->{errorCounts}->{extra} == 1, 'backup-only file recorded as extra' );
   ok( $result->{counters}->{fileCount} == 1,
      'backup-only file does NOT inflate the active-side fileCount' );
   ok( $result->{counters}->{backupOnlyCount} == 1, 'backupOnlyCount incremented' );
}
{
   my $result = runJoin( [], [], label => 'bothempty' );
   ok( $result->{joined} && scalar @{ $result->{kinds} } == 0, 'two empty listings join to nothing' );
}
{
   my $result = runJoin( [ rec( './a', 1, 1 ), rec( './b', 1, 1 ) ], [], label => 'emptybackup' );
   ok( $result->{counters}->{errorCounts}->{missing} == 2,
      'empty backup listing reports every active file missing' );
}
{
   my $result = runJoin( [], [ rec( './a', 1, 1 ), rec( './b', 1, 1 ) ], label => 'emptyactive' );
   ok( $result->{counters}->{errorCounts}->{extra} == 2,
      'empty active listing reports every backup file extra' );
}

# ---------------------------------------------------------------------------------------------
print "\n=== mergeJoinListings: the directory-name-prefix case ===\n";
# ---------------------------------------------------------------------------------------------
{
   # Correctly sorted: './Documentation.md' before './Documentation/index.html'.
   my @lines = ( rec( './Documentation.md', 50, 100 ), rec( './Documentation/index.html', 60, 200 ) );
   my $result = runJoin( [@lines], [@lines], label => 'prefix-sorted' );
   ok( join( ',', @{ $result->{kinds} } ) eq 'both,both',
      'correctly sorted prefix-colliding names pair up cleanly' );
   ok( $result->{counters}->{errorCounts}->{missing} == 0
         && $result->{counters}->{errorCounts}->{extra} == 0,
      'no phantom missing/extra findings for prefix-colliding names' );
}
{
   # The same data in BSD 'find -s' order (a directory's contents come before a sibling file
   # whose name extends the directory name). This documents the trap: the join is only correct
   # on LC_ALL=C-sorted input, and mis-sorted input yields confidently wrong findings rather
   # than an error.
   #
   # Scenario: the backup is genuinely missing ./Documentation/index.html and nothing else, so
   # the only correct finding is one 'missing'. Fed in find -s order, the join instead reports a
   # phantom 'extra' for ./Documentation.md - a file that demonstrably exists on BOTH sides.
   my $result = runJoin(
      [ rec( './Documentation/index.html', 60, 200 ), rec( './Documentation.md', 50, 100 ) ],
      [ rec( './Documentation.md', 50, 100 ) ],
      label => 'prefix-findS'
   );
   ok( $result->{counters}->{errorCounts}->{extra} > 0,
      'find -s ordering invents an "extra" for a file present on both sides (trap documented)' );
   ok( $result->{counters}->{errorCounts}->{missing} > 1,
      'find -s ordering also over-reports "missing" (trap documented)' );

   # ... and the same scenario in correct order gives exactly the one true finding.
   my $correct = runJoin(
      [ rec( './Documentation.md', 50, 100 ), rec( './Documentation/index.html', 60, 200 ) ],
      [ rec( './Documentation.md', 50, 100 ) ],
      label => 'prefix-sorted-asymmetric'
   );
   ok( $correct->{counters}->{errorCounts}->{missing} == 1
         && $correct->{counters}->{errorCounts}->{extra} == 0,
      'correctly sorted, the same scenario yields exactly one missing and no extra' );
}

# ---------------------------------------------------------------------------------------------
print "\n=== classifyPair: size and mtime comparison ===\n";
# ---------------------------------------------------------------------------------------------
{
   my $result = runJoin(
      [ rec( './f', 100, 1000 ) ],
      [ rec( './f', 999, 1000 ) ],
      label => 'sizediff'
   );
   ok( $result->{counters}->{errorCounts}->{sizeDiff} == 1, 'differing size recorded as sizeDiff' );
   ok( $result->{counters}->{filesMatched} == 0, 'a size mismatch is not counted as matched' );
   ok( $result->{counters}->{filesCompared} == 1, 'a size mismatch still counts as compared' );
   ok( $result->{findings}->[0]->{detail} =~ /active 100 vs backup 999/,
      'sizeDiff finding carries both sizes' );
}
{
   my $result = runJoin(
      [ rec( './f', 100, 1000 ) ],
      [ rec( './f', 100, 2000 ) ],
      label => 'mtimediff'
   );
   ok( $result->{counters}->{errorCounts}->{mtimeDiff} == 1, 'differing mtime recorded as mtimeDiff' );
   ok( $result->{counters}->{filesMatched} == 0, 'an mtime mismatch is not counted as matched' );
}
{
   my $result = runJoin(
      [ rec( './f', 100, 1000 ) ],
      [ rec( './f', 200, 2000 ) ],
      label => 'bothdiff'
   );
   ok( $result->{counters}->{errorCounts}->{sizeDiff} == 1
         && $result->{counters}->{errorCounts}->{mtimeDiff} == 1,
      'size and mtime both differing records both finding types' );
   ok( $result->{counters}->{filesMatched} == 0, 'double mismatch counted as matched only once (not at all)' );
}
{
   my $result = runJoin(
      [ rec( './f', 100, 1000 ) ],
      [ rec( './f', 100, 1003 ) ],
      label  => 'slack',
      mtimeSlack => 5
   );
   ok( $result->{counters}->{errorCounts}->{mtimeDiff} == 0,
      'mtime difference within mtimeSlack is tolerated' );
   ok( $result->{counters}->{filesMatched} == 1, 'file within mtimeSlack counts as matched' );
}
{
   my $result = runJoin(
      [ rec( './f', 100, 1000 ) ],
      [ rec( './f', 100, 1010 ) ],
      label  => 'slack-exceeded',
      mtimeSlack => 5
   );
   ok( $result->{counters}->{errorCounts}->{mtimeDiff} == 1,
      'mtime difference beyond mtimeSlack is still reported' );
}

# ---------------------------------------------------------------------------------------------
print "\n=== inMtimeWindow ===\n";
# ---------------------------------------------------------------------------------------------
{
   my $none = { active => 0, minMtime => undef, maxMtime => undef };
   ok( inMtimeWindow( $none, 1 ) && inMtimeWindow( $none, 999999999 ),
      'inactive window accepts everything' );

   my $newerOnly = { active => 1, minMtime => 1000, maxMtime => undef };
   ok( !inMtimeWindow( $newerOnly, 999 ), 'newer bound excludes an older file' );
   ok( inMtimeWindow( $newerOnly, 1000 ), 'newer bound is inclusive at the boundary' );
   ok( inMtimeWindow( $newerOnly, 5000 ), 'newer bound accepts a newer file' );

   my $olderOnly = { active => 1, minMtime => undef, maxMtime => 2000 };
   ok( !inMtimeWindow( $olderOnly, 2001 ), 'older bound excludes a newer file' );
   ok( inMtimeWindow( $olderOnly, 2000 ), 'older bound is inclusive at the boundary' );

   my $band = { active => 1, minMtime => 1000, maxMtime => 2000 };
   ok( !inMtimeWindow( $band, 999 ) && inMtimeWindow( $band, 1500 ) && !inMtimeWindow( $band, 2001 ),
      'a band accepts only files inside it' );
}
{
   # A file outside the window must not be counted at all - not as a file, not as a finding.
   my $result = runJoin(
      [ rec( './new', 100, 5000 ), rec( './old', 100, 500 ) ],
      [ rec( './new', 100, 5000 ) ],
      label  => 'agefilter',
      window => { active => 1, minMtime => 1000, maxMtime => undef }
   );
   ok( $result->{counters}->{fileCount} == 1,
      'age-filtered file is excluded from fileCount' );
   ok( $result->{counters}->{errorCounts}->{missing} == 0,
      'age-filtered active-only file does not produce a missing finding' );
}

# ---------------------------------------------------------------------------------------------
print "\n=== malformed record accounting ===\n";
# ---------------------------------------------------------------------------------------------
{
   my $result = runJoin(
      [ rec( './ok', 10, 100 ), "./torn-name" ],
      [ rec( './ok', 10, 100 ), "./torn-name" ],
      label => 'malformed'
   );
   ok( $result->{counters}->{listingMalformed} >= 1, 'malformed records are counted' );
   ok( $result->{counters}->{errorCounts}->{missing} == 0
         && $result->{counters}->{errorCounts}->{extra} == 0,
      'identically-fragmented names still pair, rather than flooding missing/extra' );
}

print "\nTest Summary: $passed passed, $failed failed\n";
exit( $failed == 0 ? 0 : 1 );