#!/usr/bin/perl

use Coro;
use Coro::Handle;
use Coro::Event;
use Coro::Semaphore;
use Coro::Channel;
use Coro::Signal;

use MD5;

use constant VERSION => 1;

# TODO: content-length support (HACK!)
# TODO: real message-id parsing

$|=1;

$v = 9;

$SLAVE = 1*($ARGV[0] eq "--slave");

my $ecnt;

sub slog {
   if ($_[0] <= $v) {
      print STDERR $_[1];
   }
}

# parse_folder(mbox-file-path, callback)
# callback gets called with \$header and \$body,
# $header includes the mbox From_ line without
# the leading From_ itself.
sub parse_folder {
  my ($path, $cb) = @_;

  open my $fh, "<", $path
     or die "$path: $!";

  local $/ = "\n\n";

  my ($head, $body, $offs);

  5 == read $fh, $head, 5
     or return;

  $head eq "From "
     or return;

  $offs = 0;
  while (defined ($head = <$fh>)) {
     $head =~ /^.*? [A-Z][a-z][a-z] [A-Z][a-z][a-z] [ 0-9][0-9] \d\d:\d\d:\d\d(?: [+-]\d\d\d\d)? \d\d(?:\d\d)\n/
        or die "$path: not standard mbox format header:\n$head\n";

     local $/ = "\nFrom ";
     # NEVER enable this. content-length simply is broken by design
     #if ($head =~ /^Content-Length:\s+(\d+)$/im) {
     #   $1 <= read $fh, $body, $1 + 5
     #      or die "$path: partial message in mbox";
     #} else {
        $body = <$fh>;
     #}
     chomp $body;
     $cb->($offs, \$head, \$body);
     $offs = (tell $fh) - 5;
     cede unless ++$ecnt & 1023;
  }

  1;
}

sub read_mdif {
   my ($path) = @_;
   my $fh;
   my $mdif;
   open my $fh, "<", $path
      or return { };

   defined ($_ = <$fh>)
      or die "$path: empty mdif file\n";

   do {
      if ($_ eq "[SYNCMAIL]\n") {
         while (<$fh>) {
            last unless /^([a-z]+)\s*=\s*(.*)\n$/;
            $mdif->{$1} = $2;
         }
      } elsif ($_ eq "[HOSTS]\n") {
         while (<$fh>) {
            last unless /^([^[].*)=(.*)\n$/;
            $mdif->{host}{$1} = $2;
         }
      } elsif (/^\[DIFF(\d+)\.(\d+)\]\n$/) {
         my ($gen, $mtime) = ($1, $2);
         my @dif;
         while (<$fh>) {
            last unless /^[+-]/;
            push @dif, substr $_, 0, -1;
         }
         unshift @{$mdif->{diff}}, [$gen, $mtime, \@dif];
      } elsif ($_ eq "[INDEX]\n") {
         my @idx;
         while (<$fh>) {
            last unless /^(\d+)=(.*)\n$/;
            push @idx, [$1, $2];
         }
         $mdif->{idx} = \@idx;
      } elsif (/^#/) {
         $_ = <$fh>;
         # nop
      } else {
         die "$path: unparseable section '$_'\n";
      }
   } while defined $_;

   $mdif->{version} <= VERSION
      or die "$path: version mismatch ($mdif->{version} found, <".VERSION." expected)\n";

   $mdif;
}

sub write_mdif {
   my ($path, $mdif) = @_;
   my $fh;

   open my $fh, ">", "$path~"
      or die "$path~: $!";

   print $fh "# automatically generated, do NOT edit\n";

   print $fh "[SYNCMAIL]\n";
   print $fh "$_=$mdif->{$_}\n" for (qw(fsize mtime gen version));

   print $fh "[HOSTS]\n";
   print $fh "$k=$v\n" while my ($k,$v) = each %{$mdif->{host}};

   print $fh "[INDEX]\n";
   print $fh "$_->[0]=$_->[1]\n" for @{$mdif->{idx}};

   for (reverse @{$mdif->{diff}}) {
      print $fh "[DIFF$_->[0].$_->[1]]\n";
      print $fh $_, "\n" for @{$_->[2]};
   }

   close $fh
      or die "$path~: unable to create updated .mdif: $!";

   rename "$path~", $path;
}

sub gendiff {
   my ($d1, $d2) = @_;

   my @d;
   my (%d1, %d2);

   for (@$d2) {
      undef $d2{$_->[1]};
   }

   # delete msgs in d1 but not in d2
   for (@$d1) {
      undef $d1{$_->[1]};
      push @d, "-$_->[1]" unless exists $d2{$_->[1]};
   }
   %d2 = (); # conserve memory

   # add msgs in d2 but not in d1
   for (@$d2) {
      push @d, "+$_->[1]" unless exists $d1{$_->[1]};
   }

   \@d;
}

my $check_folder = new Coro::Semaphore;

sub check_folder {
   my ($path) = @_;
   my $guard = $check_folder->guard;

   (my $conf = $path) =~ s%([^/]+$)%.$1.mdif%;

   slog 1, "checking $path\n";

   if (stat $path) {
      my ($fsize, $mtime) = (stat _)[7, 9];
      if (open my $fh, "<", $conf) {
         my %conf;
         <$fh>; # skip initial comment
         <$fh> eq "[SYNCMAIL]\n"
            or die "$conf: format error";
         while (<$fh> =~ /^([a-z]+)\s*=\s*(.*)$/) {
            $conf{$1} = $2;
         }
         return (1, \%conf) if $fsize == $conf{fsize}
                            && $mtime == $conf{mtime};
      }

      slog 2, "updating $path\n";

      my @idx;

      parse_folder $path, sub {
         my ($offs, $head, $body) = @_;
         my $mid;
         if ($$head =~ /^Message-Id:\s*(<[^<\n]+>)\s*\n/im) {
            $mid = $1;
         } else {
            $mid = MD5->hexhash("$$head\0$$body");
         }
         push @idx, [$offs, $mid];
      } or return ();

      my $mdif = read_mdif $conf;

      if ($mdif->{version}) {
         my $d = gendiff $mdif->{idx}, \@idx;
         push @{$mdif->{diff}}, [
            $mdif->{gen}++,
            $mdif->{mtime},
            $d,
         ] if @$d;
      } else {
         slog 2, "$path: new folder\n";
         $mdif->{version} ||= VERSION;
         $mdif->{gen} = 1;
      }

      $mdif->{fsize} = $fsize;
      $mdif->{mtime} = $mtime;
      $mdif->{idx}   = \@idx;

      write_mdif $conf, $mdif;

      return (2, $mdif);
   } else {
      slog 2, "$path: no longer exists\n";
      unlink $conf;

      return ();
   }
}

my $send = new Coro::Channel 10;
my $done = 0;

# request $command, $data
sub request {
   my $res;
   my $signal = new Coro::Signal;
   my $cmd = defined $_[1] ? "000+".length($_[1])." $_[0]\n$_[1]" : "000 $_[0]\n";
   $send->put([$signal, \$res, $cmd]);
   $signal->wait;
   $res;
}

# reply $id, $code, $msg, $data
sub reply {
   my $cmd = defined $_[3] ? "$_[1]+".length($_[3])." $_[2]\n$_[3]" : "$_[1] $_[2]\n";
   $send->put([undef, undef, "$_[0] $cmd"]);
}

sub handle_commands {
   my ($fh) = @_;
   my $id = "a";
   async {
      $fh->print("- 000 hello $HOSTID\n");
      while (my $r = $send->get) {
         if (defined $r->[1]) {
            my $id = ++$id;
            $request{$id} = $r;
            print STDERR "<<< $SLAVE sendign request $id:$r->[2]";#d#
            $fh->print("$id $r->[2]");
         } else {
            print STDERR "<<< $SLAVE sendign reply $r->[2]";#d#
            $fh->print($r->[2]);
         }
      }
      print STDERR "$SLAVE shutdown\n";#d#
      shutdown $fh, 1;
   };
   while (<$fh>) {
      slog 0, ">>> $SLAVE received :$_";
      /^(\S+) (\d\d\d)(?:\+(\d+))?\s*(.*)$/
         or die "protocol error, garbled command ($_)";

      my ($id, $code, $dlen, $msg) = ($1, $2, $3, $4);
      my $data;

      $fh->sysread($data, $dlen) == $dlen 
         or die "unexpected read error: $!";

      if ($code == 0) {
         if ($msg eq "quit") {
            print $fh "$id 200 quit\n";
            $send->put(undef);
            last;
         } elsif ($msg eq "nop") {
            reply $id, 200, "nop";
         } elsif ($msg =~ /^hello (.*)$/) {
            $OTHERID = $1;
            slog 3, "otherid set to $OTHERID\n";
         } else {
            die "protocol error, unknown command ($_)\n";
         }
      } else {
         my $r = delete $request{$id}
            or die "protocol error, invalid reply id ($_)\n";

         ${$r->[1]} = [$code, $msg, $data];
         $r->[0]->send;
      }

      if ($done && !%request) {
         $done = 0;
         async {
            request("quit");
            exit 0;
         };
      }
   }

   exit 0;
}

sub sync_dir {
   my ($path) = @_;

   opendir my $dh, $path
      or die "$path: $!";

   while (defined (my $folder = readdir $dh)) {
      next if $folder =~ /^\./;
      my $path = "$path/$folder";
      next unless -f $path;
      my ($status, $mdif) = check_folder $path;
      print STDERR "$path: $status | $mdif\n";#d#
      if ($SLAVE) {
         request "fhave", $folder;
      } else {
         request "fwant", $folder;
      }
   }

   $done = 1;
   my $x = request("nop");
   print STDERR "$SLAVE returned nop @$x\n";#d#
}

if ($SLAVE) {
   $HOSTID = "slave";
   async \&sync_dir, "./dst";
   handle_commands unblock \*STDIN;
} else {
   $HOSTID = "master";
   {
      use Socket;
      socketpair S1, S2, AF_UNIX, SOCK_STREAM, PF_UNSPEC;
      if (fork == 0) {
         open STDIN, "<&S2" or die;
         open STDOUT, ">&S2" or die;
         exec $0, "--slave";
         exit 255;
      }
      async \&sync_dir, "./src";
      handle_commands unblock \*S1;
   }
}