/*

  
  TODO:
     count identical data as a Validation, to deal with stupid servers.
     (saves reparsing, etc)

     cache FAILURES as well   (some timeout)
  

this treats all content and most headers as opaque

it's a packrat, keeping around old retrievals....

someone else can do pre-loading (asking for stuff that's become stale...)
(but just asking for a retrieve of everytghing loaded every
minute would do that okay, I guess.)

See
http://www.w3.org/Protocols/rfc2616/rfc2616-sec13.html#sec13



 +++ clock skew?!


 api should probably change to:
 
     retrieve_data(Address, -Retrieval, +Options)

        throws:   not_modified_since(Ret)
	          still_fresh(Ret)
		  remote_short_term_error(Ret, Err)
		  ...

     never fails.

     side effects: data in filesystem
                   Ret in rdf_db, filesystem, indexed
		   - needs many read/write locks on rdf_db -
   

     this is a slightly odd approach.  it would be clearer to
     do
         (  data_modified(Address)
	 -> retrieve_data(Address, ...)
	 ...
         )
     but http's "304 Not Modified" lets us combine these into
     one remote operation, which is worth it.
	 
*/

:- [cvs, unit_test].
cvs_id('$Id: retrieve.pl,v 1.68 2004/10/26 17:01:13 sandro Exp $').

:- ensure_loaded(library(debug)).

%%%%  Interferes with mt! :- ensure_loaded(library('semweb/rdf_db')).
:- rdf_register_ns(retr, 'http://www.w3.org/2004/03/retr/').

:- ensure_loaded(rdfpage).

:- ensure_loaded(convert).
:- ensure_loaded(time_conversions).
:- ensure_loaded(genid).
:- ensure_loaded(robot_excl).
% :- ensure_loaded(multi_retrieve).
:- ensure_loaded(retrieval_metadata).
:- ensure_loaded(shell).
:- ensure_loaded(uri).
% :- ensure_loaded(curl).


%% retrieve(+URI, +Headers, -Retrieval, +Options)
%%
%%   options:
%%     NO   queue(+Boolean)       queue it for later retrieval
%%          robot(+RobotName)     to allow for Robot Exclusion Protocol
%%          max_age(+Seconds)     override what the server says
%%          default_max_age(+Seconds)     override our guess
%%          new(-IsNewData)       used to find out if this data is new
%%
%%      Return a handle to a suitable Retrieval, which might be a new one
%%      or a good-enough recent one.  State is stored in the RDF database
%%      and maybe the file system. 
%%
%% 
%%

retrieve(URI, Headers, Retrieval, Options) :-
	(  memberchk(new(New), Options) ; true ),
	debug(retrieve, 'first look for old retrievals', []),
	(   old_retrieval(URI, Headers, Retrieval, Options)
	->  New = fail
	;   debug(retrieve, 'oh well, making a new one', []),
	    (  catch(new_retrieval(URI, Headers, Retrieval, Options),
		     my(E),	% "my" so I only catch my own errors
		     debug(retrieve, 'caught ~q', [E])
		    )
	    ->  true
	    ;   E=failed
	    ),
	    (   var(E)
	    ->  New = true
	    ;   (   E=not_modified_since(R2)
		->  debug(retrieve, 'caught ~q, ~q', [Retrieval, R2]),
		    Retrieval=R2,
		    New = fail
		    %  surely this was already true....?  rdfpage_assert(Retrieval, rdf:type, retr:'SuccessfulRetrieval')
		;   E=transient_error(Ret)
		->	rdfpage_assert(Ret, rdf:type, retr:'FailedRetrieval'),
		    Retrieval=Ret
				% short expiration
		;   E=long_term_error(Ret)
		->	rdfpage_assert(Ret, rdf:type, retr:'FailedRetrieval'),
		    Retrieval=Ret
				% long expiration
		;   throw(unknown_retrieve_error(E))
		)
	    )
	).

match(URI, [], Retrieval) :-
	!,
	(  rdfpage(Retrieval, retr:initialResource, URI)
	;  rdfpage(Retrieval, retr:finalResource, URI) ),
	rdfpage(Retrieval, rdf:type, retr:'SuccessfulRetrieval').

match(_URI, _Headers, _Retrieval) :-
	throw(headers_not_implemented).

%%  stale(+Retrieval, -Stale, +Options)
%%
%%  Options:  max_age(MaxAge)
%%
%%  Unifies Stale with the number of seconds "stale" the retrieval is;
%%  that is: how long past its expiration date we are.  Note that
%%  remote expiration time can be over-riden by the max_age option.
%%
%%  Stale > 0 mean STALE (Expired); Stale <= 0 means NOT STALE.
%%
stale(Ret, Stale, Options) :-
	age(Ret, Age),
	( % try each of these in order; first working one wins!
	    memberchk(max_age(MaxAge), Options)  
	;   rdfpage(Ret, retr:maxAgeFromServer, MaxAge)
	;   memberchk(default_max_age(MaxAge), Options)  
	;   rdfpage(Ret, retr:estimatedMaxAge, MaxAge)
	;   MaxAge is 6 * 3600   % 6 hours
	),
	!, 
	Stale is Age - MaxAge.   % negative if not stale

old_retrieval(URI, Headers, Retrieval, Options) :-
	debug(retrieve, 'Looking for a match for ~a (~q)', [URI, Headers]),
	latestCacheValidation(URI, Headers, Validation, Original),
	debug(retrieve, 'Is ~q fresh enough?', [Validation]),
	stale(Validation, Stale, Options),
	max_stale(Validation, SystemMaxStale),
	option(max_stale(MaxStale), Options, SystemMaxStale),
	% debug(retrieve, '... ~q', [[age=Age, maxAge=MaxAge, stale=Stale, maxStale=MaxStale]]),
	Stale < MaxStale,
	debug(retrieve, '... so its good enough to return as a candidate.', []),
	Retrieval = Original,
	debug(retrieve, '... returning original: ~a.', [Retrieval]).


% the first retrieval is considered a validation, here
latestCacheValidation(URI, _HeadersIgnoredOoooops, Val, Orig) :-
	% find most-recent cache validation
	findall(ret(NegTime, Validation, TimeSec, Orig),
		(
		  (   %  RFC 2616 says we MUST NOT cache the
		      %  redirect, but...   arg....  I say
		      %  we must.   Fooey.
		      rdfpage(Orig, retr:initialResource, URI)
		  
		  ;   rdfpage(Orig, retr:finalResource, URI)
		  ),
		  rdfpage(Orig, rdf:type, retr:'SuccessfulRetrieval'),
		  (  rdfpage(Orig, retr:cacheValidation, Validation)
		  ;  Orig=Validation ),
		  rdfpage(Validation, retr:transferStopTime, literal(unixtime(TimeSec))),
		  %time(TimeSec)=Time,
		  NegTime is 0 - TimeSec
		), L),
	length(L, Count),
	debug(retrieve,'~a cache validations found', [Count]),
	sort(L, LS),
	[ ret(_, Val, _TimeSec, Orig) | _ ] = LS.

latestCacheValidation(URI, Headers, Orig, Orig) :-
	match(URI, Headers, Orig),
	debug(retrieve, 'using unvalidated match', []).
	   
age(Ret, Age) :-
	% rxm(match(Ret, retr:dateServed, Born, code:retrieve)),
	ret_get(Ret, retr:exchangingHeaders, literal(unixtime(Born1))),
	debug(retrieve, 'stop_time was ~a', [Born1]),
	get_time(Now),
	% Born = time(Born1),
	Age is Now - Born1.

storedMaxAge(Ret, StoredMaxAge) :-
	rdfpage(Ret, retr:maxAgeFromServer, StoredMaxAge).
storedMaxAge(Ret, StoredMaxAge) :-
	rdfpage(Ret, retr:estimatedMaxAge, StoredMaxAge).

max_stale(_Request, 1).    % never fret about anything < 1 second




create_retrieval(URI, Ret, Options) :-
	rdfpage_gen_uri(Page),
	rdfpage_sub_uri(Page, retrieval, Ret),
	ret_set_status(Ret, retr:initializing),
	rdfpage_make_storage(Page),
	
	debug(retrieve, 'new Retrieval for ~a is ~a', [URI, Ret]),
	passert(address_latest_ret(URI, Ret)),
	passert(address_ret(URI, Ret)),
	uri_host_part(URI, Host),

	ret_set(Ret, rdf:type, retr:'Retrieval'),
	ret_set(Ret, rdfs:label, literal('A Retrieval')),
	rdfpage__group_index(Page, _Group, Index),
	atom_number(IndexAtom, Index),
	atom_concat('Data Page #', IndexAtom, Label),
	rdfpage_assert(Page, rdfs:label, literal(Label)),
	rdfpage_assert(Page, rdf:type, retr:'RDFSource'),

	ret_set(Ret, retr:host, literal(Host)),
	ret_set(Ret, retr:initialResource, URI),

	assertz(ret_watchable(Ret)),

	(   memberchk(max_age(MaxAge), Options)
	->  ret_set(Ret, retr:clientMaxAge, literal(MaxAge))
	;   true
	),

	(   memberchk(robot(RobotName), Options)
	->  ret_set(Ret, retr:robotName, literal(RobotName)),
	    (   robot_allowed(Ret)
	    ->  true
	    ;   throw(robot_excluded)
	    )
	;   true
	),

		
	% rdfpage_qsave(Page),
	log_term(retrieval, retrieval_page(Page)).

%  reconstruct the options passed to the retrieval
ret_options(Ret, Options) :-
	(   rdfpage(Ret, retr:clientMaxAge, literal(MaxAge))
	->  Options1 = [max_age(MaxAge)]
	;   Options1 = []
	),
	(   rdfpage(Ret, retr:robotName, literal(RobotName))
	->  Options2 = [robot(RobotName) | Options1]
	;   Options2 = Options1
	),
	Options = Options2.

%   Ret should have the headers & options in it.
%   [ in prolog/RDF form...?!?!? ]

ret_file(Ret, Tag, Filename) :-
	ret_page(Ret, Page),
	rdfpage_filename(Page, Tag, Filename),
	!.
	
perform_retrieval(Ret) :-

	ret_get(Ret, retr:initialResource, Address),
	debug(retrieve, 'working on retrieval ~q for ~q', [Ret, Address]),
	last_modified(Address, LM, Previous),
	accept_header(Accept),
	ret_page(Ret, Page),
	rdfpage_filename(Page, 'meta.pl', HeadersFile),
	ret_file(Ret, data, DataFile),
	prolog_download(Ret, Address, HeadersFile, DataFile, Accept, LM, Header, 8),
	add_header_metadata(Ret, Header),
	memberchk(status(StatusCode, _StatusText), Header),

	(   (   StatusCode = ok,
		ground(Previous),
		ret_page(Previous, PreviousPage),
		rdfpage_filename(PreviousPage, data, PreviousDataFile),
		ret_set_status(Ret, retr:comparingToPreviousContent),
		same_file_contents(DataFile, PreviousDataFile),
		ret_set_status(Ret, retr:doneWithComparison)
	    ;   StatusCode = not_modified
	    )
	->  ret_set(Ret, retr:sameDataAs, Previous),
	    ret_set(Ret, rdf:type, retr:'Validation'),
	    % call it successful, which means users of this need to
	    % check the sameDataAs link
	    passert(address_latest_successful_ret(Address, Ret))
	    % rdfpage_qsave(Page),
	    % throw(my(not_modified(Previous)))
	;   (   StatusCode = ok
	    ->  ret_get(Ret, retr:finalResource, FinalResource),
		ret_assert_triple(Ret, FinalResource, rdf:type, retr:'Source'),
		passert(address_latest_successful_ret(Address, Ret)),
		ret_set(Ret, rdf:type, retr:'SuccessfulRetrieval')
				% rdfpage_qsave(Page)
	    ;   ret_set(Ret, rdf:type, retr:'FailedRetrieval'),
				% rdfpage_qsave(Page),
		throw(my(http_error(StatusCode)))
	    )
	).

ret_in_progress(Ret) :-
	\+ ret_done(Ret).

ret_done(Ret) :-
	ret_get(Ret, rdf:type, retr:'SuccessfulRetrieval').
ret_done(Ret) :-
	ret_get(Ret, rdf:type, retr:'FailedRetrieval').
ret_done(Ret) :-
	ret_get(Ret, rdf:type, retr:'Validation').




	
%%   arguably we should use SHA1?
%%
%%   still, this isn't too bad, esp with the size comparison.
%%   it runs about 800KB/sec on my desktop machine.
%%
%%   ... of course diff & sha1 runs at about 100MB/sec !!
%%
%%   but I don't really like the fork.	
%%	
same_file_contents(File1, File2) :-
	size_file(File1, Size),
	size_file(File2, Size),
	open(File1, read, S1, [lock(read), buffer(full)]),
	open(File2, read, S2, [lock(read), buffer(full)]),
	catch(same_stream_contents(S1, S2),
	      not_the_same,
	      Status=fail
	     ),
	close(S1),
	close(S2),
	!,
	var(Status).   % ground -> fail -> this clause fails.


same_stream_contents(S1, S2) :-
	repeat,
	get_byte(S1, B1),
	get_byte(S2, B2),
	(  B1 = B2
	-> true
	;  throw(not_the_same)
	),
	B1 = -1.    % if not, then fail back to the repeat.

	

%%  If there was a transient error (500), and we have a Previous,
%%      - treat it like not_modified, although record it as a
%%        "implied validation"	
%%
%%  If it was a non-transient error (404), or we don't have a Previous,
%%      - record it as an Error of some sort; fail!
%%	
check_for_errors(Ret) :-
	rdf(Ret, retr:httpStatus, literal(200)), !.
check_for_errors(Ret) :-
	rdf(Ret, retr:httpStatus, literal(0)),
	!,
	throw(my(transient_error(Ret))).
check_for_errors(Ret) :-
	rdf(Ret, retr:httpStatus, literal(404)),
	!,
	throw(my(not_found(Ret))).
check_for_errors(Ret) :-
	!,
	throw(my(long_term_error(Ret))).


/*
last_modified(URI, LM, Previous) :-
	% find most-recent Retrieval
	debug(retrieve, 'lm looking for ~q', [URI]),
	findall(ret(NegTime, Retrieval),
		(
		  rdfpage(Retrieval, retr:resource , URI),
		  % only care about ones which resulted in DATA, not just validation
		  rdfpage(Retrieval, retr:status, retr:'DONE'),
		  rdfpage(Retrieval, retr:requestTime, literal(Time)),
		  unixtime(TimeSec)=Time,
		  NegTime is 1 - TimeSec,
		  debug(retrieve, 'prior @ ~w, ~q', [TimeSec, Retrieval])
		), L),
	sort(L, LS),
	(   [Head|_] = LS,
            debug(retrieve, 'head = ~q', [Head]),
            Head=ret(_, Previous),
	    debug(retrieve, 'prior chosen: ~q', [Previous]),
	    (  rdfpage(Previous, retr:lastModified, literal(LM)),
	       debug(retrieve, 'prior lm: ~q', [LM])
	    ;  rdfpage(Previous, retr:header, literal(date(LM))),
	       debug(retrieve, 'prior lm (really date): ~q', [LM])
	    )
	->  true
	;   debug(retrieve, 'lm found nothing for ~q', [URI]),
	    LM = '01 Jan 1970 00:00:01 GMT' % or blank?!    Or fail?
	).
*/

last_modified(Address, LastModifiedField, Previous) :-
	address_latest_successful_ret(Address, Previous),
	ret_get(Previous, retr:lastModifiedText, LastModifiedField),
	!.
last_modified(_, '01 Jan 1970 00:00:01 GMT', _).

	
	

/*
:- ensure_loaded(ntriples).

load_triples_from(Ret) :-
	rxm(( rdf(Ret, retr:dir, LDir, retdata) )),
	literal(Dir) = LDir,
	concat_atom([Dir, '/rapper-out.nt'], NTriplesFile),
	read_file_to_codes(NTriplesFile, Codes, []),
	ntriples_text(Triples, _BNodeMap, Codes, ""),
	length(Triples, Count),
	checklist(assertz, Triples),
	debug(retrieve, "Read ~a triples", [Count]).
*/



/*
load_into_graph(Ret, Graph) :-
	rxm(( rdf(Ret, retr:dir, LDir, retdata) )),
	literal(Dir) = LDir,
	concat_atom([Dir, '/parsed-data.pl'], File),
	(   exists_file(File)
	->  open(File, read, Stream),
	    read(Stream, Term),
	    (   Term = end_of_file
	    ->  true
	    ;   close(Stream),
		rx(anchor(Term, Graph))
	    )
	;   fail
	).

:- ensure_loaded(index).

visit(URI) :-
	retrieve(URI, [], Ret, []),
	load_into_graph(Ret, URI),
	rx(insert(Ret, retr:loadedFrom, URI, retdata)),
	rx(insert(URI, rdf:type, retr:'Source', retdata)),
	reindex,
	pre3,
	assert(loaded(URI)).
*/


% 1,184,736 inferences, 0.64 CPU in 0.79 seconds (81% CPU, 1851150 Lips)
% vs
% 29,545 inferences, 0.02 CPU in 0.02 seconds (92% CPU, 1477250 Lips)
%
%.... to consult 1500 triples.


/*
save_ret_state :-
	% BUG: write out bnodes naively.  Ooops.
	% ... oh well, not our problem here, really.  Someone else
	%     should be worrying about this.   (write in rdf/xml?)
	% rdelta_get_delta(...) maybe?
	rxe(( findall(insert(A,B,C,retdata),
		    rdf(A, B, C, retdata),
		    L)
	   )),
	length(L, Len),
	debug(retrieve, 'Saving ~a triples.', [Len]),
	open('retrieval_state.pl', write, Stream),
	format(Stream, ':- rdelta_execute((~n', []),
	write_each(L, Stream),
	format(Stream, '  noop )).', []),
	close(Stream).

write_each([], _Stream).
write_each([X|Xs], Stream) :-
	format(Stream, '  ~q,~n', [X]),
	write_each(Xs, Stream).
*/

chaals :-
	retrieve('http://www.w3.org/People/Charles/foaf', [], Ret, []),
	format('~q~n', [Ret]).

reload_cache_metadata :-        %%  OBSOLETE: see "pagein"
	% stupid, brute force way,
	% Later: have one page which is a list of the best of the others.
	write('reloading pages'),
	(   rdfpage_exists_storage(S),
	    %format('reloaded ~q~n', [S]),
	    write('.'),
	    rdfpage_load(S),
	    fail
	;   true
	), nl.

load_everything :-   
	% stupid, brute force way,
	% Later: have one page which is a list of the best of the others.
	%       rdf_reset_db,
	write('reloading pages'),
	(   rdfpage_exists_storage(S),
	    %format('reloaded ~q~n', [S]),
	    write('.'), flush,
	    rdfpage_load(S),
	    fail
	;   true
	), nl.

/*
load_retrievals :-
	with_write_lock(rdf, unsafe_load_retrievals).

unsafe_load_retrievals :-
	write('reloading list of retrieval pages'),
	retractall(retrieval_page(_)),
	(   reload_log(retrieval)
	->  nl,
	    write('reloading retrieval pages'),
	    (   retrieval_page(P),
		write('.'), flush,
		rdfpage_load(P),
		fail
	    ;   true
	    ), nl
	;   write('...   none found.'), nl
	).
*/

:- dynamic ret_asserted/4.

load_retrievals :-
	print_message(informational, format('reloading data about retrievals', [])),
	% qload if there is one....?
	retractall(ret_asserted(_,_,_,_)),
	(   reload_log(ret_asserted)
	->  flag(load_retrievals_count, _, 0),
	    (   retract(ret_asserted(S, P, O, G)),
		catch(rdf_transaction(rdf_assert(S,P,O,G)),
		      E,
		      print_message(warning, E)
		     ),
		flag(load_retrievals_count, N, N+1),
		(   M is N mod 100,
		    M = 0
		->  write('.'), flush
		;   true
		),
		fail
	    ;   true
	    ),
	    flag(load_retrievals_count, N, N),
	    print_message(informational,
			  format('~w retrieval-data triples loaded', [N]))
	;   print_message(informational,
			  format('... oh, there wasnt any.', []))
     ),
	print_message(informational, format('loading inferences...', [])),
	rdf_transaction(
			forall(ret_get(_, retr:inference, Inf),
			       pagein_graph(Inf)
			      )
		       ),
	print_message(informational, format('done.', [])).


	
	    
%   ... at some point, we can have a compiled (qsave) version
%   as a baseline, for much faster loading.   Or switch to a
%   different style rdf db.

 
% :- dynamic ret_status/3.


/*  
     "status" is a a sigma over the status change events recorded
    for the retrieval.   setting it adds an event; getting it looks
    for the most recent event.
    
*/

ret_set_status(Ret, StatusQName) :-
	get_time(Now),
	rdf_global_id(StatusQName, Status),
	(   ret_get_status(Ret, OldStatus, OldTime)
	->  (   OldStatus = Status
	    ->  true   % set-same-status; do nothing
	    ;   Dur is Now - OldTime,
		debug(retrieve, 'Ret change Status : ~w -> ~w (after ~w sec)', [OldStatus, Status, Dur]),
		ret_set(Ret, Status, literal(unixtime(Now)))
	    )
	;   debug(retrieve, 'Ret initial Status: ~w', [Status]),
	    ret_set(Ret, Status, literal(unixtime(Now)))
	).

%ret_get_status(Ret, Status, TimeStatusWasSet) :- % cached?
%	ret_status(Ret, Status, TimeStatusWasSet), !.
ret_get_status(Ret, Status, TimeStatusWasSet) :-
	findall(set(SortTime, Status),
		(   ret_get(Ret, Status, literal(unixtime(Time))),
		    % we're assuming any property which has a time
		    % value is a status...   eeek.
		    \+ rdf_equal(Status, retr:expires),
		    \+ rdf_equal(Status, retr:dateServed),
		    SortTime is -Time
		),
		Unsorted),
	sort(Unsorted, Sorted),
	Sorted = [ set(SortTime, Status) | _ ],
	TimeStatusWasSet is -SortTime.

ret_get_status_label(Ret, StatusLabel, TimeStatusWasSet) :-
	(  ret_get_status(Ret, Status, TimeStatusWasSet)
	-> once(label([], Status, StatusLabel))
	;  StatusLabel = unknown,
	   get_time(TimeStatusWasSet)
	).


%ret_get_status(_, unknown_ret_status, Now) :-
%	get_time(Now).



/*

     ret_set / ret_get gives us a front-end to rdf_db which gives
     us some more flexibility to deal with persistence and locking.

*/
     
ret_set(Ret, Predicate, Object) :-
	ret_page(Ret, Page),
	log_term(ret_asserted(Ret, Predicate, Object, Page)),
        rdf_transaction(rdf_assert(Ret, Predicate, Object, Page)),
	(   rdf(Ret, Predicate, Object, Page)
	->  true
	;   print_message(error, format('rdf_assert-then-rdf failed, ~q ~q ~q ~q', [Ret, Predicate, Object, Page]))
	).

% assert an RDF triple pertaining to the retrieval
ret_assert_triple(Ret, S, P, O) :-
	ret_page(Ret, Page),
	log_term(ret_asserted(S, P, O, Page)),
	rdf_transaction(rdf_assert(S, P, O, Page)).

ret_get(Ret, Predicate, Object) :-
	ground(Ret), !,
	ret_page(Ret, Page),
	rdf(Ret, Predicate, Object, Page).
ret_get(Ret, Predicate, Object) :-
	var(Ret), !,
	rdf(Ret, Predicate, Object, Page),
	ret_page(Ret, Page).   % leave this here to make sure we don't get untrusted stuff


ret_triple(Ret, S, P, O) :-
	ret_page(Ret, Page),
	rdf(S, P, O, Page).

:- multi_dynamic(ret_page_cache/2).   % very fast for Ret -> Page at least
ret_page(Ret, Page) :-
	ret_page_cache(Ret, Page), !.
ret_page(Ret, Page) :-
	rdfpage_sub_uri(Page, Middle, Ret),
	Middle = retrieval,
	assert(ret_page_cache(Ret, Page)).

	

	      
accept_header('application/rdf+xml,q=1; application/rdf,q=0.9; application/xml,text/xml,q=0.6; text/html,q=0.3; text/plain,q=0.2; */*,q=0.1').   % confuses emacs!

