(* $Id$
 * ----------------------------------------------------------------------
 *
 *)


(* TODO:
 * - Improve log messages
 * - Keep qlist backup
 * - Authentication
 *   Rpc_server: limit message size. Find out peer IP address, port number.
 * - Auth protocol: Password authentication
 * - Put all config options into config section.
 *)

let spooldir = "/tmp/qserver";;
let max_chunksize = Int32.of_string "1048576" ;;   (* 1 MB *)



open Netnumber
open Netxdr
open Rpc
open Rpc_server

module A = Queues_aux ;;
module S1 = Queues_srv.QUEUESPROG.QUEUESVERS1 ;;

type error =
    [ `not_found
      | `picked
      | `exists
      | `not_picked
      | `chunks_too_large
      | `timeout
      | `empty
      | `queue_deleted
      | `full
      | `bad_name
      | `bad_value
      | `inactive
      | `permission_denied
      | `sys_error
    ]

exception Error of error

let log message =
  (* Output [message] to the log medium. This is currently stderr. *)
  prerr_endline message
;;

let check_name name =
  (* Checks that [name] is a proper name. Raises Error if not. *)
  for i = 0 to String.length name - 1 do
    match name.[i] with
	'A'..'Z' -> ()
      | 'a'..'z' -> ()
      | '0'..'9' -> ()
      | '_'      -> ()
      | _        -> raise(Error `bad_name)
  done
;;


let catch_error f reply =
  try (f reply : unit)
  with
      Error code ->
	(* This match...with is only to help the typechecker inferring the
	 * right type
	 *)
	(match code with
           #error as c -> reply c
	)
    | Sys_error e ->
	log ("Sys_error: " ^ e);
	reply `sys_error
    | Unix.Unix_error(code,msg,fn) ->
	log ("Unix_error: " ^ Unix.error_message code ^
 	     (if msg <> "" then ": " ^ msg else "") ^
	     (if fn <> "" then " [" ^ fn ^ "]" else ""));
	reply `sys_error
;;


let protected_call f what =
  try
    f what
  with
      err ->
	log("Exception in deferred function: " ^
	    Printexc.to_string err);
	false    (* remove from [ongrow], [onshrink] *)
;;


let with_out_file file fn =
  try
    let r = fn file in
    close_out file;
    r
  with
      err -> close_out file; raise err
;;


let with_in_file file fn =
  try
    let r = fn file in
    close_in file;
    r
  with
      err -> close_in file; raise err
;;


let directory_list name =
  let l = ref [] in
  let dir = Unix.opendir name in
  try
    while true do
      match Unix.readdir dir with
	  "." | ".." -> ()
	| name ->
	    l := name :: !l
    done; assert false
  with
      End_of_file ->
	Unix.closedir dir;
	List.rev !l
    | err ->
	Unix.closedir dir;
	raise err
;;


let t_queue = Netxdr.validate_xdr_type A.xdrt_queue;;
let t_entry = Netxdr.validate_xdr_type A.xdrt_entry;;


type managed_entry =
    { qentry : A.entry;
      mutable is_picked : bool;
      mutable pick_connection : Rpc_server.connection_id option;
      mutable is_uploading : bool;
      mutable upload_connection : Rpc_server.connection_id option;
      mutable current_downloads : string list;   (* dhandles *)
      mutable current_upload : string option;    (* uhandle *)
    }

type managed_queue =
    { mutable qentries : managed_entry list;
      mutable next_eid : int;
      mutable qinfo : A.queue;
      mutable deleted : bool;
      mutable onshrink : (float * (bool -> bool)) list;
        (* These functions are called in order when the queue shrinks.
	 * If the functions return 'false' they are removed
	 * from the [ongrow] list.
	 * The float is the absolute time when the function should be
	 * called anyway (timeout). The argument passed to the called
	 * function indicates whether the reason of the call is a timeout (true)
	 * or a shrinking queue (false).
	 *)
      mutable ongrow : (float * (bool -> bool)) list;
        (* These functions are called in order when an entry is added to
	 * a queue. If the functions return 'false' they are removed
	 * from the [ongrow] list.
	 * The float is the absolute time when the function should be
	 * called anyway (timeout). The argument passed to the called
	 * function indicates whether the reason of the call is a timeout (true)
	 * or a growing queue (false).
	 *)
    }

type managed_download =
    { mutable downfile : in_channel;
      mutable downsize : int;
      mutable downchunksize : int;
      mutable downserial : int64;
              downentry : managed_entry;
	      downconnection : connection_id;
    }

type managed_upload =
    { mutable upfile : out_channel;
      mutable upserial : int64;
              upentry : managed_entry;
	      upconnection : connection_id;
    }


let queues = (Hashtbl.create 100 : (string, managed_queue) Hashtbl.t) ;;
  (* Cached queues *)

let next_dhandle = ref 0;;
let next_uhandle = ref 0;;

let dhandles = (Hashtbl.create 100 : (string, managed_download) Hashtbl.t) ;;
let uhandles = (Hashtbl.create 100 : (string, managed_upload) Hashtbl.t) ;;

let instance_prefix
  = Digest.string (string_of_float (Unix.gettimeofday())) ;;
  (* A prefix that is prepended before the dhandle and uhandle numbers in
   * the protocol. The instance_prefix distinguishes between instances of
   * qserver.
   *)


let queue_accepts_entries mq =
  let maxl = mq.qinfo.A.qparams.A.qmaxlen in
  mq.qinfo.A.qparams.A.qaccepting &&
  ( maxl < Int32.zero ||
    let l = List.length mq.qentries in
    (Int32.of_int l) < maxl
  )
;;


let queue_delivers_entries mq =
  mq.qinfo.A.qparams.A.qdelivering &&
  ( (* At least one unpicked entry needed: *)
    List.exists
      (fun e -> not e.is_picked && not e.is_uploading)
      mq.qentries
  )
;;


let load_managed_queue qid =
  (* Reads the "qlist" file and the "e*" files containing the meta data
   * of the entries
   *)
  let queuedir = Filename.concat spooldir qid in
  let infofile = Filename.concat queuedir "qinfo" in
  let listfile = Filename.concat queuedir "qlist" in
  if not (Sys.file_exists infofile) then raise (Error `not_found);
  let qinfo =
    with_in_file (open_in_bin infofile)
      (fun f ->
	 let l = in_channel_length f in
	 let buf = String.create l in
	 really_input f buf 0 l;
	 A._to_queue (Netxdr.unpack_xdr_value ~fast:true buf t_queue [])
      )
  in
  let eid_list =
    with_in_file (open_in listfile)
      (fun f ->
	 let l = ref [] in
	 try
	   while true do
	     let line = input_line f in
	     l := (int_of_string line) :: !l;
	   done;
	   assert false
	 with
	     End_of_file -> List.rev !l
      )
  in
  let next_eid =
    (List.fold_left max 0 eid_list) + 1 in
  let mentry_list =
    List.map
      (fun eid ->
	 let qentry_file =
	   Filename.concat queuedir ("e" ^ string_of_int eid) in
	 let v =
	   with_in_file (open_in_bin qentry_file)
	     (fun f ->
		let l = in_channel_length f in
		let buf = String.create l in
		really_input f buf 0 l;
		buf
	     )
	 in
	 let e =
	   A._to_entry (Netxdr.unpack_xdr_value ~fast:true v t_entry []) in
	 { qentry = e;
	   is_picked = false;
	   pick_connection = None;
	   is_uploading = false;
	   upload_connection = None;
	   current_downloads = [];
	   current_upload = None
	 }
      )
      eid_list
  in
  let mqueue =
    { qentries = mentry_list;
      next_eid = next_eid;
      deleted = false;
      qinfo = qinfo;
      onshrink = [];
      ongrow = [];
    }
  in
  mqueue
;;


let store_managed_queue qid mq =
  (* Writes the "qlist" file. Does not write anything else. *)
  let queuedir = Filename.concat spooldir qid in
  let infofile = Filename.concat queuedir "qinfo" in
  let listfile = Filename.concat queuedir "qlist" in
  if not (Sys.file_exists infofile) then raise (Error `not_found);
  (* Update qmodified: *)
  mq.qinfo.A.qmodified <- Int64.of_float (Unix.time());
  with_out_file (open_out_bin infofile)
    (fun f ->
       let v = A._of_queue mq.qinfo in
       Netxdr.pack_xdr_value v t_queue [] (output_string f);
    );
  with_out_file (open_out listfile)
    (fun f ->
       List.iter
	 (fun me ->
	    (* Write now the "qlist" file. Don't include entries currently
	     * being uploaded because these are considered as non-persistent
	     * until the upload finishes.
	     * Picked entries are considered as still persistent, so include
	     * them.
	     *)
	    if not me.is_uploading then begin
	      output_string f me.qentry.A.eid;
	      output_string f "\n";
	    end
	 )
	 mq.qentries;
    )
;;


let lookup_managed_queue qid =
  (* Checks whether the queue is cached. If yes, the cached queue record
   * is returned. Otherwise, the queue is loaded and added to the cache.
   *)
  try
    Hashtbl.find queues qid
  with
      Not_found ->
	let mq = load_managed_queue qid in
	Hashtbl.add queues qid mq;
	(* Note: There are not any picked and uploaded entries, so reset: *)
	mq.qinfo.A.qpicked <- Int32.zero;
	mq.qinfo.A.quploads <- Int32.zero;
	(* Check the length: *)
	let l = Int32.of_int (List.length mq.qentries) in
	if l <> mq.qinfo.A.qlength then begin
	  log "Load error: The qlength field is not correct. (fixed)";
	  mq.qinfo.A.qlength <- l;
	end;
	mq
;;


let proc_ping session () reply =
  reply()
;;

let proc_create_queue session queuename =
  catch_error
    (fun reply ->
       let user = Rpc_server.get_user session in
       (* Sanity check: queuename must only consist of alphanumeric characters,
	* including underscore
	*)
       check_name queuename;
       let queuedir = Filename.concat spooldir queuename in
       let infofile = Filename.concat queuedir "qinfo" in
       let listfile = Filename.concat queuedir "qlist" in
       (* Check if the queue already exists: *)
       if Sys.file_exists infofile then raise(Error `exists);
       (* Create the directory [queuedir] if necessary: *)
       if not (Sys.file_exists queuedir) then
	 Unix.mkdir queuedir 0o777;
       (* Create qlist: *)
       with_out_file (open_out listfile)
	 (fun f -> () );
       (* Create qinfo: *)
       let now = Int64.of_float (Unix.time()) in
       let qinfo = { A.qid = queuename;
		     A.qname = queuename;
		     A.qowner = user;
		     A.qcreation = now;
		     A.qmodified = now;
		     A.qlength = Int32.zero;
		     A.qpicked = Int32.zero;
		     A.quploads = Int32.zero;
		     A.qparams = { A.qmaxlen = Int32.minus_one;
				   A.qactive = false;
				   A.qaccepting = true;
				   A.qdelivering = true;
				 }
		   } in
       with_out_file (open_out_bin infofile)
	 (fun f ->
	    let v = A._of_queue qinfo in
	    Netxdr.pack_xdr_value v t_queue [] (output_string f);
	 );
       reply `successful
    )
;;


let proc_delete_queue session queuename =
  catch_error
    (fun reply ->
       let user = Rpc_server.get_user session in
       (* Sanity check: queuename must only consist of alphanumeric characters,
	* including underscore
	*)
       check_name queuename;
       let queuedir = Filename.concat spooldir queuename in
       (* Remove the files: *)
       let files = directory_list queuedir in
       List.iter
	 (fun f -> let path = Filename.concat queuedir f in Sys.remove path)
	 files;
       (* Remove the directory. It is ok if this fails (no permission) *)
       ( try Unix.rmdir queuedir with _ -> ());
       (* Is there a managed_queue? *)
       begin try
	 let mq = Hashtbl.find queues queuename in  (* or Not_found *)
	 if mq.qinfo.A.qowner <> user then raise(Error `permission_denied);
	 Hashtbl.remove queues queuename;
	 mq.deleted <- true;
	 (* Remove all upload/download handles: *)
	 List.iter
	   (fun me ->
	      List.iter
		(Hashtbl.remove dhandles)
		me.current_downloads;
	      ( match me.current_upload with
		    None -> ()
		  | Some uh -> Hashtbl.remove uhandles uh
	      );
	   )
	   mq.qentries;
	 (* Tell the [ongrow] and [onshrink] handlers that something
	  * errorneous happened
	  *)
	 List.iter
	   (fun (_,f) -> ignore(protected_call f true)) mq.ongrow;
	 List.iter
	   (fun (_,f) -> ignore(protected_call f true)) mq.onshrink;
	 mq.ongrow <- [];
	 mq.onshrink <- [];
       with
	   Not_found -> ()   (* no managed_queue *)
       end;
       reply `successful
    )
;;


let proc_list_queues session () =
  Printf.eprintf "LIST_QUEUES\n%!";
  catch_error
    (fun reply ->
       (* Get all files in the spooldir: *)
       let l = directory_list spooldir in
       (* Keep only files that are actually directories containing "qinfo": *)
       let l' =
	 List.filter
	   (fun name ->
	      Sys.file_exists
		(Filename.concat
		   (Filename.concat spooldir name)
		   "qinfo")
	   )
	   l in
       (* Read in the "qinfo" files, and decode them: *)
       let qlist =
	 List.map
	   (fun name ->
	      let mq = lookup_managed_queue name in
	      mq.qinfo
	   )
	   l'
       in
       (* Pass the result back: *)
       reply (`successful (Array.of_list qlist))
    )
;;


let proc_get_queue session queuename =
  catch_error
    (fun reply ->
       (* Sanity check: queuename must only consist of alphanumeric characters,
	* including underscore
	*)
       check_name queuename;
       let mq = lookup_managed_queue queuename in
       reply (`successful mq.qinfo);
    )
;;


let proc_set_queue session (queuename,params) =
  catch_error
    (fun reply ->
       let user = Rpc_server.get_user session in
       (* Sanity check: queuename must only consist of alphanumeric characters,
	* including underscore
	*)
       check_name queuename;
       let mq = lookup_managed_queue queuename in
       if mq.qinfo.A.qowner <> user then raise(Error `permission_denied);
       if params.A.qactive then begin
	 (* Activate the queue/change limits: *)
	 let old_accepting = queue_accepts_entries mq in
	 let old_delivering = queue_delivers_entries mq in
	 mq.qinfo.A.qparams <- params;
	 let new_accepting = queue_accepts_entries mq in
	 let new_delivering = queue_delivers_entries mq in
	 store_managed_queue queuename mq;
	 (* NOTE: If the queue has previously been inactive, the [ongrow] and
	  * [onshrink] lists are empty. So the following statements do
	  * nothing.
	  *)
	 (* Maybe there is now space for growth: *)
	 if not old_delivering && new_delivering then
	   mq.ongrow <-
             List.filter (fun (_,f) -> protected_call f false) mq.ongrow;
	 (* Maybe there is now an entry to pick: *)
	 if not old_accepting && new_accepting then
	   mq.onshrink <-
             List.filter (fun (_,f) -> protected_call f false) mq.onshrink;
       end
       else begin
	 (* Deactivate the queue: *)
	 mq.qinfo.A.qparams <- params;
	 store_managed_queue queuename mq;
	 (* Make that everybody notices the inactive queue: *)
	 mq.ongrow <-
           List.filter (fun (_,f) -> protected_call f true) mq.ongrow;
	 mq.onshrink <-
           List.filter (fun (_,f) -> protected_call f true) mq.onshrink;
	 (* Make these lists empty: *)
	 if mq.ongrow <> [] then
	   log "Warning: After deactivation the ongrow list is not empty";
	 if mq.onshrink <> [] then
	   log "Warning: After deactivation the onshrink list is not empty";
	 mq.ongrow <- [];
	 mq.onshrink <- [];
       end;
       reply `successful
    )
;;


let proc_list_queue_entries session qid =
  catch_error
    (fun reply ->
       let user = Rpc_server.get_user session in
       (* Sanity check: qid must only consist of alphanumeric characters,
	* including underscore
	*)
       check_name qid;
       let mq = lookup_managed_queue qid in
       if mq.qinfo.A.qowner <> user then raise(Error `permission_denied);
       let visible =
	 List.filter
	   (fun qe ->
	      not qe.is_picked && not qe.is_uploading
	   )
	   mq.qentries
       in
       let visible' = List.map (fun qe -> qe.qentry) visible in
       reply (`successful (Array.of_list visible'))
    )
;;


let timeout_pick mq =  (* raise the right exception *)
  if mq.deleted then
    raise (Error `queue_deleted)
  else
    if not mq.qinfo.A.qparams.A.qactive then
      raise (Error `inactive)
  else
    if mq.qinfo.A.qparams.A.qdelivering then
      raise (Error `empty)
    else
      raise (Error `timeout)
;;


let try_pick session mq reply0 timed_out =
  let keep = ref true in
  let reply r =
    (* If [reply] is called, the function [try_pick] must return [false]
     * to indicate the it is to be removed from the [ongrow] list
     *)
    keep := false; reply0 r
  in
  catch_error
    (fun reply ->
       if timed_out then timeout_pick mq;
       if not mq.qinfo.A.qparams.A.qactive then raise(Error `inactive);
       try
	 if not mq.qinfo.A.qparams.A.qdelivering then raise Not_found;
	 let first =
	   List.find
	     (fun qe -> not qe.is_picked && not qe.is_uploading)
	     mq.qentries
	 in (* or Not_found *)
	 reply (`successful first.qentry);  (* may raise Connection_lost *)
	 first.is_picked <- true;
	 first.pick_connection <- Some(Rpc_server.get_connection_id session);
	 mq.qinfo.A.qpicked <- Int32.succ mq.qinfo.A.qpicked;
	 mq.qinfo.A.qlength <- Int32.pred mq.qinfo.A.qlength;
       with
	   Not_found -> ()
	       (* Try again *)
    )
    reply;
  !keep




let proc_pick_queue_entry session (qid,timeout) =
  catch_error
    (fun reply ->
       let user = Rpc_server.get_user session in
       (* Sanity check: qid must only consist of alphanumeric characters,
	* including underscore
	*)
       check_name qid;
       let mq = lookup_managed_queue qid in
       if mq.qinfo.A.qowner <> user then raise(Error `permission_denied);
       if try_pick session mq reply false then begin
	 (* [try_pick] was not successful. Add [try_pick] to the [ongrow]
	  * list
	  *)
	 if timeout = Int32.zero then timeout_pick mq;
	 let max_time =
	   if timeout >= Int32.zero then
	     Unix.time() +. Int32.to_float timeout
           else
	     (-1.0)
	 in
	 mq.ongrow <- mq.ongrow @ [ max_time, try_pick session mq reply ];
       end
    )
;;


let stop_downloads mq entry =
  List.iter
    (fun dhandle ->
       let dl =
	 try Hashtbl.find dhandles dhandle
	 with Not_found -> assert false
       in
       close_in dl.downfile;
       Hashtbl.remove dhandles dhandle
    )
    entry.current_downloads;
  mq.qinfo.A.qpicked <- Int32.pred mq.qinfo.A.qpicked;
  mq.qinfo.A.qlength <- Int32.succ mq.qinfo.A.qlength;
  entry.current_downloads <- []
;;


let proc_return_picked_queue_entry session (qid,eid) =
  catch_error
    (fun reply ->
       let user = Rpc_server.get_user session in
       (* Sanity check: qid must only consist of alphanumeric characters,
	* including underscore
	*)
       check_name qid;
       let mq = lookup_managed_queue qid in
       if mq.qinfo.A.qowner <> user then raise(Error `permission_denied);
       let entry =
	 try
	   List.find
	     (fun qe -> qe.qentry.A.eid = eid)
	     mq.qentries
	 with Not_found -> raise (Error `not_found)
       in

       if not entry.is_picked then raise (Error `not_picked );
       if entry.pick_connection <> Some(Rpc_server.get_connection_id session)
       then raise(Error `not_found);

       entry.is_picked <- false;
       (* Stop all current downloads: *)
       stop_downloads mq entry;
       reply `successful
    )
;;


let proc_remove_picked_queue_entry session (qid,eid) =
  catch_error
    (fun reply ->
       let user = Rpc_server.get_user session in
       (* Sanity check: qid must only consist of alphanumeric characters,
	* including underscore
	*)
       check_name qid;
       let mq = lookup_managed_queue qid in
       if mq.qinfo.A.qowner <> user then raise(Error `permission_denied);
       let entry =
	 try
	   List.find
	     (fun qe -> qe.qentry.A.eid = eid)
	     mq.qentries
	 with Not_found -> raise (Error `not_found)
       in

       if not entry.is_picked then raise (Error `not_picked );
       if entry.pick_connection <> Some(Rpc_server.get_connection_id session)
       then raise(Error `not_found);

       (* Stop all current downloads: *)
       stop_downloads mq entry;
       (* Remove this entry from the list, and store the list: *)
       mq.qentries <- List.filter
	                (fun qe -> qe.qentry.A.eid <> eid)
	                mq.qentries;
       mq.qinfo.A.qlength <- Int32.pred mq.qinfo.A.qlength;
       store_managed_queue qid mq;
       (* Remove the "e*" and "d*" files: *)
       let queuedir = Filename.concat spooldir qid in
       let qentry_file = Filename.concat queuedir ("e" ^ eid) in
       let qdata_file = Filename.concat queuedir ("d" ^ eid) in
       Sys.remove qentry_file;
       Sys.remove qdata_file;
       reply `successful;
       (* Finally, consider the [onshrink] activities: *)
       mq.onshrink <- List.filter
	                (fun (_,f) -> protected_call f false) mq.onshrink
    )
;;


let proc_remove_queue_entry session (qid,eid) =
  catch_error
    (fun reply ->
       let user = Rpc_server.get_user session in
       (* Sanity check: qid must only consist of alphanumeric characters,
	* including underscore
	*)
       check_name qid;
       let mq = lookup_managed_queue qid in
       if mq.qinfo.A.qowner <> user then raise(Error `permission_denied);
       let entry =
	 try
	   List.find
	     (fun qe -> qe.qentry.A.eid = eid)
	     mq.qentries
	 with Not_found -> raise (Error `not_found)
       in

       if entry.is_picked &&
	  entry.pick_connection <> Some(Rpc_server.get_connection_id session)
       then raise(Error `not_found);
       if entry.is_picked then raise (Error `picked );
       if entry.is_uploading then raise (Error `sys_error);  (* strange *)

       (* Remove this entry from the list, and store the list: *)
       mq.qentries <- List.filter
	                (fun qe -> qe.qentry.A.eid <> eid)
	                mq.qentries;
       mq.qinfo.A.qlength <- Int32.pred mq.qinfo.A.qlength;
       store_managed_queue qid mq;
       (* Remove the "e*" and "d*" files: *)
       let queuedir = Filename.concat spooldir qid in
       let qentry_file = Filename.concat queuedir ("e" ^ eid) in
       let qdata_file = Filename.concat queuedir ("d" ^ eid) in
       Sys.remove qentry_file;
       Sys.remove qdata_file;
       reply `successful;
       (* Finally, consider the [onshrink] activities: *)
       mq.onshrink <- List.filter
	                (fun (_,f) -> protected_call f false) mq.onshrink
    )
;;


let proc_download_entry session (qid,eid,chunksize) =
  catch_error
    (fun reply ->
       let user = Rpc_server.get_user session in
       (* Sanity check: qid must only consist of alphanumeric characters,
	* including underscore
	*)
       check_name qid;
       if chunksize > max_chunksize then raise (Error `chunks_too_large);
       if chunksize <= Int32.zero then raise (Error `bad_value);
       let ichunksize = Int32.to_int chunksize in
       let mq = lookup_managed_queue qid in
       if mq.qinfo.A.qowner <> user then raise(Error `permission_denied);
       let entry =
	 try
	   List.find
	     (fun qe -> qe.qentry.A.eid = eid)
	     mq.qentries
	 with Not_found -> raise (Error `not_found)
       in

       if not entry.is_picked then raise (Error `not_picked );
       if entry.pick_connection <> Some(Rpc_server.get_connection_id session)
       then raise(Error `not_found);

       (* Create new dhandle: *)
       let n = !next_dhandle in
       incr next_dhandle;
       let dhandle =
	 instance_prefix ^ Netnumber.BE.int4_as_string (int4_of_int n) in
       (* Create new download record *)
       let queuedir = Filename.concat spooldir qid in
       let qdata_file = Filename.concat queuedir ("d" ^ eid) in
       let file = open_in_bin qdata_file in
       begin try
	 let dl =
	   { downfile = file;
	     downsize = in_channel_length file;
	     downchunksize = ichunksize;
	     downserial = Int64.zero;
	     downentry = entry;
	     downconnection = Rpc_server.get_connection_id session;
	   }
	 in
	 Hashtbl.add dhandles dhandle dl;
	 entry.current_downloads <- dhandle :: entry.current_downloads;
       with
	   err -> close_in file; raise err
       end;
       (* Note: [file] remains open. It will be closed once the download
	* is finished, or the download stops for some reason.
	*)
       (* Return the dhandle: *)
       reply (`successful dhandle)
    )
;;


let proc_download_chunk session dhandle =
  catch_error
    (fun reply ->
       let dl =
	 try Hashtbl.find dhandles dhandle
	 with Not_found -> raise (Error `not_found)
       in
       if dl.downconnection <> Rpc_server.get_connection_id session
       then raise(Error `not_found);

       let buf = String.create (dl.downchunksize) in
       let n = input dl.downfile buf 0 dl.downchunksize in
       let chunk =
	 { A.serial = dl.downserial;
	   A.last = pos_in dl.downfile >= dl.downsize;
	   A.data =
	     if n < dl.downchunksize then
	       String.sub buf 0 n
	     else
	       buf;
	 }
       in
       dl.downserial <- Int64.succ dl.downserial;
       reply (`successful chunk)
    )
;;


let timeout_upload mq =  (* raise the right exception *)
  if mq.deleted then
    raise (Error `queue_deleted)
  else
    if not mq.qinfo.A.qparams.A.qactive then
      raise (Error `inactive)
  else
    if mq.qinfo.A.qparams.A.qaccepting then
      raise (Error `full)
    else
      raise (Error `timeout)
;;


let stop_upload mq entry =
  match entry.current_upload with
      None -> ()
    | Some uhandle ->
	let ul =
	  try Hashtbl.find uhandles uhandle
	  with Not_found -> assert false
	in
	close_out ul.upfile;
	let queuedir = Filename.concat spooldir entry.qentry.A.eqid in
	let qdata_file = Filename.concat queuedir ("d" ^ entry.qentry.A.eid) in
	Sys.remove qdata_file;
	Hashtbl.remove uhandles uhandle;
	mq.qinfo.A.quploads <- Int32.pred mq.qinfo.A.quploads;
	entry.current_upload <- None
;;


let start_upload session mq props reply0 timed_out =
  let keep = ref true in
  let reply r =
    (* If [reply] is called, the function [start_upload] must return [false]
     * to indicate the it is to be removed from the [onshrink] list
     *)
    keep := false; reply0 r
  in
  catch_error
    (fun reply ->
       if timed_out then timeout_upload mq;
       if not mq.qinfo.A.qparams.A.qactive then raise (Error `inactive);
       (* Is there space in the queue? *)
       let qid = mq.qinfo.A.qid in
       if queue_accepts_entries mq then begin
	 (* Create new uhandle: *)
	 let n = !next_uhandle in
	 incr next_uhandle;
	 let uhandle =
	   instance_prefix ^ Netnumber.BE.int4_as_string (int4_of_int n) in
	 (* Create new entry record in memory: *)
	 let eid = string_of_int (mq.next_eid) in
	 mq.next_eid <- mq.next_eid + 1;
	 let qe =
	   { A.eid = eid;
	     A.eqid = qid;
	     A.ecreation = Int64.of_float (Unix.time());
	     A.esize = Int64.zero;
	     A.eprops = props;
	   }
	 in
	 let entry =
	   { qentry = qe;
	     is_picked = false;
	     pick_connection = None;
	     is_uploading = true;
	     upload_connection = Some (Rpc_server.get_connection_id session);
	     current_downloads = [];
	     current_upload = Some uhandle;
	   }
	 in
	 (* Create new data file: *)
	 let queuedir = Filename.concat spooldir qid in
	 let qdata_file = Filename.concat queuedir ("d" ^ eid) in
	 let file = open_out_bin qdata_file in
	 (* Note: the entry file will be created when the upload is finished *)
	 begin try
	   (* Create new upload record: *)
	   let ul =
	     { upfile = file;
	       upserial = Int64.zero;
	       upentry = entry;
	       upconnection = Rpc_server.get_connection_id session;
	     }
	   in
	   (* Register [entry]: *)
	   mq.qentries <- mq.qentries @ [ entry ];
	   mq.qinfo.A.quploads <- Int32.succ mq.qinfo.A.quploads;
	   (* Register [ul]: *)
	   Hashtbl.add uhandles uhandle ul;
	 with
	     err -> close_out file; raise err
	 end;
	 (* The upload is accepted: *)
	 try
	   reply (`successful uhandle);
	 with
	     Rpc_server.Connection_lost ->
	       stop_upload mq entry
       end
    )
    reply;
  !keep
;;


let proc_upload_entry session (qid,props,timeout) =
  catch_error
    (fun reply ->
       let user = Rpc_server.get_user session in
       (* Sanity check: qid must only consist of alphanumeric characters,
	* including underscore
	*)
       check_name qid;
       (* If the queue is short enough, begin the upload immediately;
	* otherwise wait
	*)
       let mq = lookup_managed_queue qid in
       if mq.qinfo.A.qowner <> user then raise(Error `permission_denied);
       if start_upload session mq props reply false then begin
	 (* Upload is not possible now. Check if should do it later. *)
	 if timeout = Int32.zero then raise (Error `timeout);
	 let max_time = if timeout >= Int32.zero then
	                   Unix.time() +. Int32.to_float timeout
	                else
			  -1.0
	 in
	 mq.onshrink <- mq.onshrink @
	                [ max_time, start_upload session mq props reply ];
       end
    )
;;


let proc_upload_chunk session (uhandle,chunk) =
  catch_error
    (fun reply ->
       let ul =
	 try Hashtbl.find uhandles uhandle
	 with Not_found -> raise (Error `not_found)
       in
       if ul.upconnection <> Rpc_server.get_connection_id session
       then raise(Error `not_found);

       if chunk.A.serial <> ul.upserial then raise (Error `bad_value);
       output_string ul.upfile chunk.A.data;
       let l = String.length chunk.A.data in
       ul.upentry.qentry.A.esize <-
         Int64.add ul.upentry.qentry.A.esize (Int64.of_int l);
       if chunk.A.last then begin
	 let entry = ul.upentry in
	 (* Close the data file *)
	 close_out ul.upfile;
	 (* Write the entry file *)
	 let qid = entry.qentry.A.eqid in
	 let eid = entry.qentry.A.eid in
	 let queuedir = Filename.concat spooldir qid in
	 let qentry_file = Filename.concat queuedir ("e" ^ eid) in
	 with_out_file (open_out_bin qentry_file)
	   (fun file ->
	      let v = A._of_entry entry.qentry in
	      Netxdr.pack_xdr_value v t_entry [] (output_string file);
	   );
	 (* Deallocate uhandle: *)
	 Hashtbl.remove uhandles uhandle;
	 entry.current_upload <- None;
	 (* Make the new entry visible: *)
	 let mq = lookup_managed_queue qid in
	 entry.is_uploading <- false;
	 mq.qinfo.A.qlength <- Int32.succ mq.qinfo.A.qlength;
	 mq.qinfo.A.quploads <- Int32.pred mq.qinfo.A.quploads;
	 store_managed_queue qid mq;
	 (* Tell waiters that there is a new entry: *)
	 mq.ongrow <- List.filter
	                (fun (_,f) -> protected_call f false) mq.ongrow;
	 reply `successful
       end
       else begin
	 ul.upserial <- Int64.succ ul.upserial;
	 reply `successful
       end
    )
;;


let onclose conn_id =
  (* This function is called when a TCP connection is closed. We have to
   * iterate over the whole cache of queues and to:
   * - return all picked entries (by this connection)
   * - stop all downloads (by this connection)
   * - stop and remove all uploads (by this connection)
   *)
  Hashtbl.iter
    (fun qid mq ->
       let canceled_uploads = ref false in
       List.iter
	 (fun entry ->
	    if entry.is_picked && entry.pick_connection = Some conn_id then
	      begin
		entry.is_picked <- false;
		stop_downloads mq entry;
	      end;

	    if entry.is_uploading && entry.upload_connection = Some conn_id then
	      begin
		entry.is_uploading <- false;
		stop_upload mq entry;
		canceled_uploads := true;
	      end;
	 )
	 mq.qentries;
       (* If there are cancaled uploads, execute [onshrink]: *)
       if !canceled_uploads then
	 mq.onshrink <- List.filter
	                  (fun (_,f) -> protected_call f false) mq.onshrink;
    )
    queues
;;


let check_timeouts ev =
  (* This function is called every minute. We have to
   * iterate over the whole cache of queues and to:
   * - return all timed out picked entries and stop all downloads
   * - stop and remove all timed out uploads
   *)
  ( match ev with
	Unixqueue.Timeout(_,_) -> ()
      | _ -> raise Equeue.Reject
  );
  let now = Unix.time() in
  Hashtbl.iter
    (fun qid mq ->
       (* Are there timed out events in [onshrink]? *)
       mq.onshrink <-
         List.filter
	   (fun (t,f) ->
	      if t >= 0.0 && t <= now then
		protected_call f true
	      else
		true
	   )
	   mq.onshrink;
       (* Are there timed out events in [ongrow]? *)
       mq.ongrow <-
         List.filter
	   (fun (t,f) ->
	      if t >= 0.0 && t <= now then
		protected_call f true
	      else
		true
	   )
	   mq.ongrow;
    )
    queues
;;

let pluggable_auth_module = 
  ref ("<None>", 
       (`Socket(Tcp, Rpc_server.Portmapped, Rpc_server.default_socket_config)),
       (fun _ -> 
	  failwith "No auth module linked, startup not possible"; () )) ;;


let main() =
  Printexc.record_backtrace true;
  let esys = Unixqueue.create_unix_event_system() in

  let (auth_name, srv_mode, f_srv_config) = !pluggable_auth_module in
  prerr_endline ("Starting queues server with auth module: " ^ auth_name);

(*
  Rpc_client.verbose true;
  Rpc_server.verbose true;

  Rpc_client.Debug.enable_ptrace := true;
 *)

  let server = Rpc_server.create2 srv_mode esys in

  S1.bind_async
    ~proc_ping
    ~proc_create_queue
    ~proc_delete_queue
    ~proc_list_queues
    ~proc_get_queue
    ~proc_set_queue
    ~proc_list_queue_entries
    ~proc_pick_queue_entry
    ~proc_return_picked_queue_entry
    ~proc_remove_picked_queue_entry
    ~proc_remove_queue_entry
    ~proc_download_entry
    ~proc_download_chunk
    ~proc_upload_entry
    ~proc_upload_chunk
    server;

  Rpc_server.set_onclose_action server onclose;

  f_srv_config server;

  (* Arrange that [check_timeouts] is called every second: *)
  let w = Unixqueue.new_wait_id esys in
  let g = Unixqueue.new_group esys in
  Unixqueue.add_handler esys g (fun _ _ -> check_timeouts);
  Unixqueue.add_resource esys g (Unixqueue.Wait w, 1.0);

  List.iter
    (fun signal ->
       Netsys_signal.register_handler
	 ~signal
	 ~name:"Qserver"
	 ~callback:(fun _ ->
		      Unixqueue.remove_resource esys g (Unixqueue.Wait w);
		      Rpc_server.stop_server server;
		   )
	 ()
    )
    [ Sys.sighup; Sys.sigint; Sys.sigquit; Sys.sigterm ];

  Sys.set_signal
    Sys.sigpipe
    Sys.Signal_ignore;

  let rec auto_restart f arg =
    try f arg
    with err ->
      prerr_endline ("Server: Uncaught exception: " ^ Printexc.to_string err);
      auto_restart f arg
  in

  (* Fork *)
  match Unix.fork() with
      0 ->
        (* Child *)
	Sys.chdir "/";
	ignore(Unix.setsid());
        auto_restart Unixqueue.run esys;
        exit 99
    | n when n > 0 ->
        (* Parent *)
	()
    | _ ->
        assert false

;;