dtRdr::Annotation::Sync::Standard - standard server sync

  my $sync = dtRdr::Annotation::Sync::Standard->new($uri);

  $sync->start;

  $sync->work;

  $sync->finish;

  ($user, $pass) = $self->authenticate($realm, $uri);

  $self->add_request($method, $path);

  $self->queue($token => sub {...});

  my @subs = $self->dequeue($token);

  $self->next_request($token);

  my $cookies = $self->cookies;

  $self->process_config(%data);

  $self->process_manifest(%data);

  $self->incoming($id, %answer);

  $self->adjust_times('out', $obj);

  $self->adjust_times('in', $obj, $have);

  $book = $self->book($anno);

package dtRdr::Annotation::Sync::Standard;
$VERSION = eval{require version}?version::qv($_):$_ for(0.0.1);

use warnings;
use strict;
use Carp;

use dtRdr::UserAgent;
use dtRdr::Annotation::IOBlob;
use dtRdr::Annotation::SyncRules;

use HTTP::Status; # XXX RC_THBBT
use YAML::Syck ();
my $LOAD = sub {my $v = YAML::Syck::Load(@_);
  return(defined($v) ? $v : {})};
my $DUMP = sub {YAML::Syck::Dump(@_)};

use dtRdr::Logger;

use constant NOISE => 0;

use base 'MultiTask::Minion';

use Class::Accessor::Classy;
ro 'anno_io';
ro 'ua';
ro 'uri';
ro 'server';
ro 'bookbag';
ri 'cookies';
rw 'auth_sub';
rw 'server_drift';
lw 'books';
# TODO cookie_something?
# TODO progress_sub?
no  Class::Accessor::Classy;

sub new {
  my $package = shift;
  my $class = ref($package) || $package;
  my $uri;
  $uri = shift(@_) if(@_ % 2);
  my %args = @_;
  $uri = $args{server}->uri unless(defined($uri));

  $uri =~ s#/*$#/#;

  my $self = {
    %args,
    uri => $uri,
    _queue => {},
  };
  bless($self, $class);
  return($self);
} # end subroutine new definition
########################################################################

sub start {
  my $self = shift;

  my $ua = $self->{ua} ||= dtRdr::UserAgent->new;
  $ua->set_complete_sub(sub {shift; $self->next_request(@_)});
  $ua->set_auth_sub(sub {shift; $self->authenticate(@_)});

  # TODO start with login/etc?

  # setup the work chain, starting with the time sync on config.yml
  my $token = $self->add_request('GET', 'config.yml');
  $self->queue($token => sub {$self->process_config(%{$_[1]})});

} # end subroutine start definition
########################################################################

sub work {
  my $self = shift;

  NOISE and warn "working\n";

  $self->ua->pester and $self->finish;

} # end subroutine work definition
########################################################################

sub finish {
  my $self = shift;

  # check for leftovers
  my %left = $self->ua->leftovers;
  foreach my $token (keys(%left)) {
    die "failed to finish $token ", $left{$token}{response}->code;
  }
  $self->quit;
} # end subroutine finish definition
########################################################################

sub authenticate {
  my $self = shift;
  my ($realm, $uri) = @_;

  # The credential cache will look like [$token, [$u,$p], [$u,$p]].
  # The second $u,$p is to make the cache logic quicker and disallow the
  # same server wanting a different password for different pages.
  my $c = $self->{_auth_cache} ||= [];

  # hmm, should we allow one $server object to have different realms?
  # Probably too silly, and the config would have to support that.

  # return the cached credential unless we're stuck on the same page
  if(scalar(@$c) == 3) {
    return(@{$c->[2]});
  }

  my $s = $self->server;

  # make a token (those shouldn't be undef, but ...)
  my $token = join("|", map({defined($_) ? $_ : '~'} $uri, $realm));

  # more cache logic
  if(@$c and ($c->[0] ne $token)) { # must have worked last time, we moved on
    push(@$c, $c->[1]); # mark it as golden
    return(@{$c->[2]});
  }

  # now we're either at the first try or some failed auth

  my $callback = sub {
    my $cb = $self->auth_sub or die "no auth_sub callback";
    my @ans = $cb->($s, $realm, $uri);
    $self->quit unless(@ans);
    return(@ans);
  };

  unless(@$c) { # first try
    # try the user's input if we have it
    my ($u, $p) = ($s->username, $s->password);
    unless(defined($u) and defined($p)) {
      ($u, $p) = $callback->();
    }
    @$c = ($token, [$u, $p]);
    return($u,$p);
  }

  # TODO If we fix LWP, we have to configurably limit the number of
  # times we nag to support non-gui usage.

  # now we were just plain wrong
  my ($u, $p) = $callback->();

  return(@{$c->[1]} = ($u, $p));
} # end subroutine authenticate definition
########################################################################

sub add_request {
  my $self = shift;
  my ($method, $path, @and) = @_;

  return($self->ua->add_request($method, $self->uri . $path, @and));
} # end subroutine add_request definition
########################################################################

sub queue {
  my $self = shift;
  my ($token, $subref) = @_;

  my $Q = $self->{_queue};
  $Q->{$token} ||= [];
  push(@{$Q->{$token}}, $subref);
} # end subroutine queue definition
########################################################################

sub dequeue {
  my $self = shift;
  my ($token) = @_;

  my $Q = $self->{_queue};
  exists($Q->{$token}) or return;
  return(@{$Q->{$token}});
} # end subroutine dequeue definition
########################################################################

sub next_request {
  my $self = shift;
  my ($token) = @_;

  my %result = $self->ua->collect($token);
  foreach my $subref ($self->dequeue($token)) {
    $subref->($token, \%result);
  }
} # end subroutine next_request definition
########################################################################

sub cookies {
  my $self = shift;

  if(my $c = $self->SUPER::cookies) {
    return($c);
  }

  require HTTP::Cookies;
  my $cookies = HTTP::Cookies->new(
    # TODO do we have any need to save cookies?
    #file       => "/tmp/cookies.txt", autosave   => 1,
  );
  $self->ua->cookie_jar($cookies);
  $self->set_cookies($cookies);
} # end subroutine cookies definition
########################################################################

sub process_config {
  my $self = shift;
  my (%result) = @_;

  # regardless of the result, we should have a time_shift value
  $self->set_server_drift($result{time_shift} || 0);

  my $resp = $result{response};
  my $conf = {};
  if(RC_OK == $resp->code) {
    $conf = $LOAD->($result{string});
  }

  # slightly odd callback control-flow:
  # setup the GET_manifest sequence, then either login+this or just this
  my @books = $self->books or die "need books to sync";
  my $after_login = sub {
    my $arg = '?' . join('&', map({'book=' . $_} @books));
    my $token = $self->add_request('GET', 'manifest.yml'.$arg);
    $self->queue($token => sub {$self->process_manifest(%{$_[1]})});
  };

  # TODO properly deal with auth_required: if unset, we don't do this
  # bit if we're only reading.  For now, auth_required=1
  if(my $lconf = $conf->{login}) {
    $self->cookies;
    my $url = $lconf->{url} or die "need url for login directive";
    my $template = $lconf->{template} or die "need template for login";
    my ($u,$p) = $self->authenticate($self->server->id, $url);
    #$template =~ s/\s*$//;
    $template =~ s/#USERNAME#/$u/ or die "no #USERNAME# in template?";
    $template =~ s/#PASSWORD#/$p/ or die "no #PASSWORD# in template?";
    my $token = $self->ua->add_request('POST', $url,
      [Content_Type => "application/x-www-form-urlencoded"], $template);
    $self->queue($token => sub {
      my ($t, $result) = @_;
      my $res = $result->{response};
      my $code = $res->code;
      NOISE and warn "login response:  ", $res->as_string;
      ($code == RC_OK) or die "failed login '$code' ", $res->message;
      NOISE and warn "we logged in";
      $after_login->();
    });
  }
  else {
    $after_login->();
  }

} # end subroutine process_config definition
########################################################################

sub process_manifest {
  my $self = shift;
  my (%result) = @_;

  # TODO maybe allow a modified-since scheme too

  NOISE and warn "process_manifest\n";

  my $yaml = $result{string};
  my $resp = $result{response};
  if(RC_OK != $resp->code) {
    die "(", $resp->code, ") server not happy: $yaml";
  }

  my %s_man = %{$LOAD->($yaml)};
  # be super-strict about the manifest
  foreach my $id (keys(%s_man)) {
    defined($s_man{$id}) or
      die "server sent undefined revision for $id";
  }

  # any SELECT involved on the server needs to be mirrored here
  # for now just based on $self->books and the server id
  my $sid = $self->server->id;
  my $OBlob = sub {
    # TODO put this in the io req?
    my %book_ok = map({$_ => 1} $self->books);
    grep({$book_ok{$_->book}}
      grep({my $p; $p = $_->public and ($p->server eq $sid)}
        map({dtRdr::Annotation::IOBlob->outgoing(%$_)} @_)
      )
    )
  };

  my %current = map({$_->id => $_} $OBlob->($self->anno_io->items));
  my %deleted = map({$_->id => $_} $OBlob->($self->anno_io->deleted));

  ASSERT: { # deletes are sane
    foreach my $id (keys(%deleted)) {
      exists($current{$id}) and die "'$id' got undeleted incorrectly";
      defined($deleted{$id}->public->owner) and
        die "'$id' was deleted, but does not belong to me";
    }
  }
  ASSERT: { # prev's are sane
    foreach my $id (keys(%s_man)) {
      my $anno = $current{$id} || $deleted{$id} or next;
      ($anno->public->rev <= $s_man{$id}) or
        die "'$id' has higher synchronized revision than server!";
    }
  }

  my %actions;
  %actions = (
    REMOTE_DELETE => sub {
      my ($anno) = @_;
      my $id = $anno->id;
      my $rev = $anno->public->rev;
      my $token = $self->add_request(
        'DELETE',
        'annotation/' . $id . ".yml?rev=$rev",
      );
      $self->queue($token => sub {
        ($_[1]->{response}->code == RC_OK) or
          die YAML::Syck::Dump([@_]);
        # TODO it will send us the yaml of what gets deleted.
        # We should check it.
        $self->anno_io->x_finish_delete($anno->id);
      });
    },
    LOCAL_DELETE => sub {
      my ($anno) = @_;
      NOISE and warn "delete locally: ", $anno->id;
      $self->anno_io->s_delete($anno->id, $self->book($anno));
    },
    CONFLICT_DELETE => sub { # TODO warn/prompt?
      $actions{LOCAL_DELETE}->(@_);
    },
    POST => sub {
      my ($anno) = @_;
      my $id = $anno->id;
      my $send = $self->adjust_times('out', $anno);
      my $token = $self->add_request(
        'POST',
        'annotation/',
        [content_type => 'text/x-yaml'],
        $DUMP->($send->deref),
      );
      $self->queue($token => sub {
        my $code = $_[1]->{response}->code;
        ($code == RC_CREATED) or die "bad return code '$code',",
          "expected CREATED", YAML::Syck::Dump([@_]);
        $anno->public->set_rev($anno->revision);
        $self->anno_io->s_update($id, $anno->deref, $self->book($anno));
      });
    },
    PUT  => sub {
      my ($anno) = @_;
      my $id = $anno->id;
      my $rev = $anno->public->rev;
      my $send = $self->adjust_times('out', $anno);
      my $token = $self->add_request(
        'PUT',
        'annotation/' . $id . ".yml?rev=$rev",
        [content_type => 'text/x-yaml'],
        $DUMP->($send->deref),
      );
      $self->queue($token => sub {
        ($_[1]->{response}->code == RC_OK) or
          die YAML::Syck::Dump([@_]);
        $anno->public->set_rev($anno->revision);
        $self->anno_io->s_update($id, $anno->deref, $self->book($anno));
      });
    },
    FRESHEN => sub { # like get, but just an update from the server
      my ($anno) = @_;
      my $id = $anno->id;
      my $token =
        $self->add_request('GET', 'annotation/' . $id . '.yml');
      $self->queue($token => sub {
        $self->incoming($id, %{$_[1]},
          expect => $s_man{$id},
          have   => $anno,
        );
      });
    },
    GET  => sub {
      my ($id) = @_; # no object here
      my $token =
        $self->add_request('GET', 'annotation/' . $id . '.yml');
      $self->queue($token => sub {
        $self->incoming($id, %{$_[1]}, expect => $s_man{$id});
      });
    },
    SKIP => 0,
  );

  my $sr = dtRdr::Annotation::SyncRules->new(
    current  => [values(%current)],
    deleted  => [values(%deleted)],
    manifest => {%s_man},
    #error    => sub {
    #  my ($anno, @message) = @_;
    #  return('DIED.' . $message[0], $anno);
    #},
  )->init;
  while(my @ans = $sr->next) {
    my ($action, $anno) = @ans;
    exists($actions{$action}) or
      die "nothing for '$action' (" . (eval {$anno->id} || $anno) . ')';
    my $run = $actions{$action} or next;
    $run->($anno);
  }

} # end subroutine process_manifest definition
########################################################################

sub incoming {
  my $self = shift;
  my ($id, %result) = @_;
  NOISE and warn "incoming $id";

  my $yaml = $result{string};
  my $resp = $result{response};
  if(RC_OK != $resp->code) {
    die "(", $resp->code, ") for $id server not happy: $yaml ";
  }

  my $got = $LOAD->($yaml);
  $got = dtRdr::Annotation::IOBlob->incoming(%$got);
  ($got->id eq $id) or die "server sent the wrong id for '$id'";

  my $rev = $got->revision;
  defined($rev) or
    die "server sent no (or undefined) revision for '$id'";
  if(defined(my $exp = $result{expect})) {
    ($exp == $rev) or # TODO $self->complain ?
      NOISE and warn "revision was not what I expected";
  }

  $got = $self->adjust_times('in', $got, $result{have} || undef);
  # set the public attribs
  my $server = $self->server or croak("need a server object here");
  my $p = $got->public;

  # these two are spoof-prevention
  $p->set_rev($rev);
  $p->set_server($server->id);

  # set 'me' as undef
  #   XXX note this could be an edge case when username changes on the
  #   server without a local update -- force login?
  $p->set_owner(undef) if($p->owner eq $server->username);

  # TODO do we need to dig for race conditions where the annotation has
  # changed locally since the sync started?

  # explicitly distinguish update from get
  my $io = $self->anno_io;
  my $method = 's_insert';
  if(my $anno = $result{have}) {
    # TODO just do an inner update on the object?
    $method = 's_update';
  }
  $io->$method($id, $got->deref, $self->book($got));
} # end subroutine incoming definition
########################################################################

sub adjust_times {
  my $self = shift;
  my ($mode, $obj, $have) = @_;

  $obj = $obj->clone;

  my $dir = {out => 1, in => -1}->{$mode} or
    croak("'$mode' is not 'in' or 'out'");

  my $drift = $self->server_drift;
  $drift or return($obj); # XXX what about undef fixups?

  $drift *= $dir;
  my @attribs = qw(create_time mod_time);
  if(($dir < 0) and $have) { # incoming freshen
    # don't change create time if we already have it
    # TODO check drift value?
    shift(@attribs);
  }
  foreach my $att (@attribs) {
    my $v = $obj->$att;
    defined($v) or next;
    my $set_att = 'set_' . $att;
    $obj->$set_att($v + $drift);
  }
  return($obj);
} # end subroutine adjust_times definition
########################################################################

sub book {
  my $self = shift;
  my ($anno) = @_;

  my $bag = $self->bookbag or return;

  my $bid = eval {$anno->book};
  $@ and ($bid = $anno);
  return($bag->find($bid));
} # end subroutine book definition
########################################################################

# vi:ts=2:sw=2:et:sta
1;