Megatest

;;     GNU General Public License for more details.
;; 
;;     You should have received a copy of the GNU General Public License
;;     along with Megatest.  If not, see <http://www.gnu.org/licenses/>.

;;======================================================================





(declare (unit rmtmod))

(declare (uses apimod))
(declare (uses commonmod))
(declare (uses configfmod))
(declare (uses dbmod))
(declare (uses debugprint))
(declare (uses itemsmod))
(declare (uses mtargs))
(declare (uses mtver))
(declare (uses pgdb))
(declare (uses portloggermod))
(declare (uses servermod))
(declare (uses tasksmod))


(module rmtmod
	*
	
(import scheme
		
	chicken.base

	directory-utils
	format
	;; http-client
	;; intarweb
	matchable
	md5
	message-digest
	nng ;; nanomsg
	(prefix base64 base64:)
	(prefix sqlite3 sqlite3:)
	regex
	s11n
	;; spiffy
	;; spiffy-directory-listing
	;; spiffy-request-vars

	pgdb
	pkts
	portloggermod
	(prefix mtargs args:)
	servermod
	stml2
	tasksmod
	)

(defstruct alldat
  (areapath #f)
  (ulexdat  #f)
  )


;; (require-extension (srfi 18) extras tcp s11n)
;; 
;; 
;; (use  srfi-1 posix regex regex-case srfi-69 hostinfo md5 message-digest posix-extras)
;; 
;; (use spiffy uri-common intarweb http-client spiffy-request-vars intarweb spiffy-directory-listing)
;; 
;; Configurations for server
;; (tcp-buffer-size 2048)
;; (max-connections 2048) 

;; info about me as a server

;;
(defstruct servdat
  (host #f)
  (port #f)
  (uuid #f)
  (rep  #f)
  (dbfile #f)
  (api-url #f)
  (api-uri #f)
  (api-req #f)
  (status 'starting)
  (trynum 0) ;; count the number of ports we've tried

  ) 



(define (servdat->url sdat)
  (conc (servdat-host sdat)":"(servdat-port sdat)))

;; generate entries for ~/.megatestrc with the following
;;
;;  grep define ../rmt.scm | grep rmt: |perl -pi -e 's/\(define\s+\((\S+)\W.*$/\1/'|sort -u

(defstruct remotedat
  (conns (make-hash-table)) ;; apath/dbname => conndat
  )

(defstruct conndat
  (apath    #f)
  (dbname   #f)
  (fullname #f)
  (hostport #f)
  (ipaddr   #f)
  (port     #f)
  (socket   #f)
  (srvpkt   #f)
  (srvkey   #f)
  (lastmsg  0)
  (expires  0)
  (inport   #f)
  (outport  #f))

(define *srvpktspec*
  `((server (host    . h)
	    (port    . p)
	    (servkey . k)
	    (pid     . i)
	    (ipaddr  . a)
	    (dbpath  . d))))

;;======================================================================
;;  S U P P O R T   F U N C T I O N S
;;======================================================================

;; replaces *runremote*
(define *remotedat* (make-remotedat))

;; -> http://abc.com:900/<entrypoint>
;;
(define (conndat->uri conn entrypoint)
  (conc "http://"(conndat-ipaddr conn)":"(conndat-port conn)"/"entrypoint))

;; set up the api proc, seems like there should be a better place for this?

(define api-proc (make-parameter conc))
(api-proc api:process-request)

;; do we have a connection to apath dbname and
;; is it not expired? then return it
;;
;; else setup a connection
;;
;; if that fails, return '(#f "some reason") ;; NB// convert to raising an exception
;;
(define (rmt:get-conn remdat apath dbname)
  (let* ((fullname (db:dbname->path apath dbname))
	 (conn     (hash-table-ref/default (remotedat-conns remdat) fullname #f)))
    (if (and conn
	     (< (current-seconds) (conndat-expires conn)))
	conn
	#f)))


(define (rmt:find-main-server apath dbname)
  (let* ((pktsdir     (get-pkts-dir apath))
	 (all-srvpkts (get-all-server-pkts pktsdir *srvpktspec*))
	 ;; (dbpath      (conc apath "/" dbname))
	 (viable-srvs (get-viable-servers all-srvpkts dbname)))
    (get-the-server apath viable-srvs)))


(define *connstart-mutex* (make-mutex))
(define *last-main-start* 0)

;; looks for a connection to main, returns if have and not exired
;; creates new otherwise
;; 
;; connections for other servers happens by requesting from main
;;
;; TODO: This is unnecessarily re-creating the record in the hash table
;;
(define (rmt:open-main-connection remdat apath)
  (let* ((fullpath (db:dbname->path apath "/.db/main.db"))
	 (conns    (remotedat-conns remdat))
	 (conn     (hash-table-ref/default conns fullpath #f))) ;; TODO - create call for this















    (cond



     ((and conn                                             ;; conn is NOT a socket, just saying ...
	   (< (current-seconds) (conndat-expires conn)))
      #t) ;; we are current and good to go - we'll deal elsewhere with a server that was killed or died 
     ((and conn
	   (>= (current-seconds)(conndat-expires conn)))
      (debug:print-info 0 *default-log-port* "connection to "fullpath" server expired. Reconnecting.")
      (if (conndat-socket conn)
	  (nng-close! (conndat-socket conn)))
      (hash-table-set! conns fullpath #f) ;; clean up
      (rmt:open-main-connection remdat apath))
     (else
      ;; Below we will find or create and connect to main

      (let* ((dbname         (db:run-id->dbname #f))
	     (the-srv        (rmt:find-main-server apath dbname))
	     (start-main-srv (lambda () ;; call IF there is no the-srv found
			       (mutex-lock! *connstart-mutex*)
			       (if (> (- (current-seconds) *last-main-start*) 5) ;; at least four seconds since last attempt to start main server
				   (begin
				     (api:run-server-process apath dbname)
				     (set! *last-main-start* (current-seconds))
				     (thread-sleep! 1)))

			       (mutex-unlock! *connstart-mutex*)
			       (rmt:open-main-connection remdat apath) ;; TODO: Add limit to number of tries
			       )))
	(if (not the-srv) ;; have server, try connecting to it
	    (start-main-srv)
	    (let* ((srv-addr (server-address the-srv)) ;; need serv
		   (ipaddr   (alist-ref 'ipaddr  the-srv))
		   (port     (alist-ref 'port    the-srv))
		   (srvkey   (alist-ref 'servkey the-srv))
		   (fullpath (db:dbname->path apath dbname))
		   
		   (new-the-srv (make-conndat
				 apath:   apath
				 dbname:  dbname
				 fullname: fullpath
				 hostport: srv-addr
				 socket: (open-nn-connection srv-addr)
				 ipaddr: ipaddr
				 port: port
				 srvpkt: the-srv
				 srvkey: srvkey ;; generated by rmt:get-signature on the server side
				 lastmsg: (current-seconds)
				 expires: (+ (current-seconds) 60) ;; this needs to be gathered during the ping


				 )))
	      (hash-table-set! conns fullpath new-the-srv)))
	#t)))))

;; NB// remdat is a remotedat struct
;;
(define (rmt:general-open-connection remdat apath dbname #!key (num-tries 5))
  (assert (not (equal? dbname ".db/main.db")) "ERROR: general-open-connection should never be called with main as the db")
  (let* ((mdbname  (db:run-id->dbname #f))
	 (fullname (db:dbname->path apath dbname))
	 (conns    (remotedat-conns remdat))
	 (mconn    (rmt:get-conn remdat apath mdbname)))

    (if (and mconn
	     (not (debug:print-logger)))
	(begin
	  (debug:print-info 0 *default-log-port* "Turning on logging to main, look in logs dir for main log.")
	  (debug:print-logger rmt:log-to-main)))
    (cond




     ((or (not mconn) ;; no channel open to main?
	  (< (conndat-expires mconn)(+ (current-seconds) 2))) ;; restablish connection if less than 2 seconds on the lease
      (if mconn ;; previously opened - clean up NB// consolidate this with the similar code in open main above
	  (begin
	    (debug:print-info 0 *default-log-port* "Clearing out connection to main that has expired.")
	    (nng-close! (conndat-socket mconn))
	    (hash-table-set! conns fullname #f)))
      (rmt:open-main-connection remdat apath)
      (rmt:general-open-connection remdat apath mdbname))
     ((not (rmt:get-conn remdat apath dbname))                 ;; no channel open to dbname?     
      (let* ((res (rmt:send-receive-real remdat apath mdbname 'get-server `(,apath ,dbname))))
	(case res
	  ((server-started)
	   (if (> num-tries 0)
	       (begin
		 (thread-sleep! 2)
		 (rmt:general-open-connection remdat apath dbname num-tries: (- num-tries 1)))
	       (begin
		 (debug:print-error 0 *default-log-port* "Failed to start servers needed or open channel to "apath", "dbname)
		 (exit 1))))
	  (else
	   (if (list? res) ;; server has been registered and the info was returned. pass it on.
	       (begin ;;  ("192.168.0.9" 53817
		      ;;  "5e34239f48e8973b3813221e54701a01" "24310"
		      ;;  "192.168.0.9"
		      ;;  "/home/matt/data/megatest/tests/simplerun"
		 ;;  ".db/1.db")
		 (match
		  res
		  ((host port servkey pid ipaddr apath dbname)
		   (debug:print-info 0 *default-log-port* "got "res)
		   (hash-table-set! conns
				    fullname
				    (make-conndat
				     apath: apath
				     dbname: dbname
				     hostport: (conc host":"port)
				     socket: (open-nn-connection (conc host":"port))
				     ipaddr: ipaddr
				     port: port
				     srvkey: servkey
				     lastmsg: (current-seconds)
				     expires: (+ (current-seconds) 60))))


		  (else
		   (debug:print-info 0 *default-log-port* "return data from starting server did not match host port servkey pid ipaddr apath dbname " res)))
		 res)
	       (begin
		 (debug:print-info 0 *default-log-port* "Unexpected result: " res)
		 res)))))))
    
    #t))

;;======================================================================

;; FOR DEBUGGING SET TO #t
;; (define *localmode* #t)
(define *localmode* #f)
(define *dbstruct* (make-dbr:dbstruct))

;; Defaults to current area
;;
(define (rmt:send-receive cmd rid params #!key (attemptnum 1)(area-dat #f))
  ;; (if (not *remotedat*)(set! *remotedat* (make-remotedat)))
  (let* ((apath      *toppath*)
	 (remdat     *remotedat*)
	 (conns      (remotedat-conns remdat)) ;; just checking that remdat is a remotedat
	 (dbname     (db:run-id->dbname rid)))
    (if *localmode*
	(let* ((dbdat    (dbr:dbstruct-get-dbdat *dbstruct* dbname))
	       (indat    `((cmd . ,cmd)(params . ,params))))
	  (api:process-request *dbstruct* indat)
	  ;; (api:process-request dbdat indat)
	  )
	(begin
	  (rmt:open-main-connection remdat apath)
	  (if rid (rmt:general-open-connection remdat apath dbname))
	  (rmt:send-receive-real remdat apath dbname cmd params)))))


#;(define (rmt:send-receive-setup conn)
  (if (not (conndat-inport conn))
      (let-values (((i o) (tcp-connect (conndat-ipaddr conn)
				       (conndat-port conn))))
	(conndat-inport-set! conn i)
	(conndat-outport-set! conn o))))
  
;; db is at apath/.db/dbname, rid is an intermediary solution and will be removed
;; sometime in the future
;;
(define (rmt:send-receive-real remdat apath dbname cmd params)

  (let* ((conn (rmt:get-conn remdat apath dbname)))
    (assert conn "FATAL: rmt:send-receive-real called without the needed channels opened")

    (assert (conndat-socket conn) "FATAL: rmt:send-receive-real called without the channel socket opened.")
    (let* ((soc     (conndat-socket conn))
	   (key     #f)
	   (host    (conndat-ipaddr conn))
	   (port    (conndat-port   conn))
	   (payload `((cmd    . ,cmd)
		      (key    . ,(conndat-srvkey conn))
		      (params . ,params)))
	   (res      (send-receive-nn soc ;; (open-send-receive-nn (conc host":"port)




				      (sexpr->string payload))))
      (if (member res '("#<unspecified>")) ;; TODO - fix this in string->sexpr
	  #f

	  (string->sexpr res)))))

;; db is at apath/.db/dbname, rid is an intermediary solution and will be removed
;; sometime in the future.
;;
;; Purpose - call the main.db server and request a server be started
;; for the given area path and dbname
;;
;; (define (rmt:send-receive-server-start remdat apath dbname)
;;   (let* ((conn (rmt:get-conn remdat apath dbname)))
;;     (assert conn "FATAL: Unable to connect to db "apath"/"dbname)
;;     #;(let* ((res      (with-input-from-request
;; 		      (conndat->uri conn "api") 
;; 		      `((params . (,apath ,dbname)))
;; 		      read-string)))
;;       (string->sexpr res))))

(define (rmt:print-db-stats)
  (let ((fmtstr "~40a~7-d~9-d~20,2-f")) ;; "~20,2-f"
    (debug:print 18 *default-log-port* "DB Stats, "(seconds->year-week/day-time (current-seconds))"\n=====================")
    (debug:print 18 *default-log-port* (format #f "~40a~8a~10a~10a" "Cmd" "Count" "TotTime" "Avg"))
    (for-each (lambda (cmd)
		(let ((cmd-dat (hash-table-ref *db-stats* cmd)))

(define (rmt:kill-server run-id)
  (rmt:send-receive 'kill-server #f (list run-id)))

(define (rmt:start-server run-id)
  (rmt:send-receive 'start-server #f (list run-id)))

(define (rmt:get-server-info apath dbname)
  (rmt:send-receive 'get-server-info #f (list apath dbname)))

;;======================================================================
;;  M I S C
;;======================================================================

(define (rmt:login run-id)

(define (common:api-changed?)
  (not (equal? (substring (->string megatest-version) 0 4)
               (substring (conc (common:get-last-run-version)) 0 4))))

;; host and port are used to ensure we are remove proper records
(define (rmt:server-shutdown host port)
  (let ((dbfile   (servdat-dbfile *server-info*)))
    (debug:print-info 0 *default-log-port* "dbfile is "dbfile)
    (if dbfile
	(let* ((am-server  (args:get-arg "-server"))
	       (dbfile     (args:get-arg "-db"))
	       (apath      *toppath*)
	       (remdat     *remotedat*)) ;; foundation for future fix
	  (if *dbstruct-db*
	      (let* ((dbdat      (db:get-dbdat *dbstruct-db* apath dbfile))
		     (db         (dbr:dbdat-db dbdat))
		     (inmem      (dbr:dbdat-db dbdat))
		     )
		;; do a final sync here
		(debug:print-info 0 *default-log-port* "Doing final sync for "apath" "dbfile" at "(current-seconds))
		(db:sync-inmem->disk *dbstruct-db* apath dbfile force-sync: #t)
		;; let's finalize here
		(debug:print-info 0 *default-log-port* "Finalizing db and inmem")
		(if (sqlite3:database? db)
		    (sqlite3:finalize! db)
		    (debug:print-info 0 *default-log-port* "in rmt:server-shutdown, db is not a database, not finalizing..."))
		(if (sqlite3:database? inmem)
		    (sqlite3:finalize! inmem)
		    (debug:print-info 0 *default-log-port* "in rmt:server-shutdown, inmem is not a database, not finalizing..."))
		(debug:print-info 0 *default-log-port* "Finalizing db and inmem complete"))
	      (debug:print-info 0 *default-log-port* "Db was never opened, no cleanup to do."))
	  (if (not am-server)
	      (debug:print-info 0 *default-log-port* "I am not a server, should NOT get here!")
	      (if (string-match ".*/main.db$" dbfile)
		  (let ((pkt-file (conc (get-pkts-dir *toppath*)
					"/" (servdat-uuid *server-info*)
					".pkt")))
		    (debug:print-info 0 *default-log-port* "removing pkt "pkt-file)
		    (delete-file* pkt-file)
		    (debug:print-info 0 *default-log-port* "Releasing lock (if any) for "dbfile ", host "host", port "port)
		    (db:with-lock-db
		     (servdat-dbfile *server-info*)
		     (lambda (dbh dbfile)
		       (db:release-lock dbh dbfile host port)))) ;; I'm not the server - should not have a lock to remove
		  (let* ((sdat *server-info*) ;; we have a run-id server
			 (host (servdat-host sdat))
			 (port (servdat-port sdat))
			 (uuid (servdat-uuid sdat))
			 (res  (rmt:deregister-server remdat *toppath* host port uuid dbfile)))
		    (debug:print-info 0 *default-log-port* "deregistered-server, res="res)
		    (debug:print-info 0 *default-log-port* "deregistering server "host":"port" with uuid "uuid)
		    )))))))

(define (std-exit-procedure)
  ;;(common:telemetry-log-close)
  (on-exit (lambda () 0))
  ;;(debug:print-info 13 *default-log-port* "std-exit-procedure called; *time-to-exit*="*time-to-exit*)
  (let ((no-hurry  (if (bdat-time-to-exit *bdat*) ;; hurry up
		       #f
		       (begin
			 (bdat-time-to-exit-set! *bdat* #t)
			 #t))))
    (debug:print-info 4 *default-log-port* "starting exit process, finalizing databases.")
    (if (and no-hurry (debug:debug-mode 18))
	(rmt:print-db-stats))
    (let ((th1 (make-thread
		(lambda () ;; thread for cleaning up, give it five seconds
		  (let* ((start-time (current-seconds)))
		    (if (and *server-info*
			     *unclean-shutdown*)
			(begin
			  (debug:print-info 0 *default-log-port* "Unclean server exit, calling server-shtudown")



			  (rmt:server-shutdown (servdat-host *server-info*)

					       (servdat-port *server-info*))))
		    (debug:print-info 0 *default-log-port* "Shutdown activities completed in "(- (current-seconds) start-time)" seconds"))
		  ;; (if *dbstruct-db* (db:close-all *dbstruct-db*)) ;; one second allocated
		  #;(if (bdat-task-db *bdat*)    ;; TODO: Check that this is correct for task db
		  (let ((db (cdr (bdat-task-db *bdat*))))
		  (if (sqlite3:database? db)
		  (begin
		  (debug:print-info 0 *default-log-port* "Closing down task db "db)

;; called in megatest.scm, host-port is string hostname:port
;;
;; NOTE: This is NOT called directly from clients as not all transports support a client running
;;       in the same process as the server. 
;;
;; conn is a conndat record
;;
(define (server:ping conn #!key (do-exit #f))
  (let* ((req (conndat-socket conn))
	 (srvkey (conndat-srvkey conn))
	 (msg (sexpr->string '(ping ,srvkey))))
    (send-receive-nn req msg))) ;; (server-ready? host port server-id))

;;======================================================================
;; http-transportmod.scm contents moved here
;;======================================================================

(define (http-transport:make-server-url hostport)
  (if (not hostport)
      #f
      (conc "http://" (car hostport) ":" (cadr hostport))))

;;======================================================================
;; S E R V E R
;; ======================================================================

(define (http-get-function fnkey)
  (hash-table-ref/default *http-functions* fnkey (lambda () "nothing here yet")))




;; Main entry point to start a server. was start-server
(define (rmt:run hostn)








  ;;  ;; Configurations for server
  ;;  (tcp-buffer-size 2048)
  ;;  (max-connections 2048) 
  (debug:print 2 *default-log-port* "PID: "(current-process-id)". Attempting to start the server ...")
  (let* ((db              #f) ;;        (open-db)) ;; we don't want the server to be opening and closing the db unnecesarily
	 (hostname        (get-host-name))
	 (ipaddrstr       (let ((ipstr (if (string=? "-" hostn)
					   ;; (string-intersperse (map number->string (u8vector->list (hostname->ip hostname))) ".")
					   (server:get-best-guess-address hostname)

					   #f)))
			    (if ipstr ipstr hostn))) ;; hostname))) 
	 (port            (portlogger:open-run-close portlogger:find-port))
	 ;; (link-tree-path  (common:get-linktree))
	 ;; (tmp-area        (common:get-db-tmp-area))
	 #;(start-file      (conc tmp-area "/.server-start")))
    (debug:print-info 0 *default-log-port* "portlogger recommended port: " port)
    (if *server-info*
	(begin
	  (servdat-host-set! *server-info* ipaddrstr)
	  (servdat-port-set! *server-info* port)
	  (servdat-status-set! *server-info* 'trying-port)
	  (servdat-trynum-set! *server-info* (+ (servdat-trynum *server-info*) 1)))
	(set! *server-info* (make-servdat host: ipaddrstr port: port)))
    (let* ((rep (rmt:try-start-server ipaddrstr port)))
      (let loop ((instr (nng-recv rep)))
	(let* ((data   (string->sexpr instr))
	       (res    (case data
			 ((quit) 'quit)
			 (else    (api:process-request *dbstruct-db* data))))
	       (resdat (sexpr->string res)))
	  (if (not (eq? res 'quit))
	      (begin
		(set! *db-last-access* (current-seconds))
		(nng-send rep resdat)
		(loop (nng-recv rep)))))))
    (debug:print-info 0 *default-log-port* "After server, should never see this")
    ;; server exit stuff here
    (let* ((portnum (servdat-port *server-info*))
	   (host    (servdat-host *server-info*)))
      (portlogger:open-run-close portlogger:set-port portnum "released")
      (if (not (equal? (get-host-name) host))
	  (debug:print-info 0 *default-log-port* "Server shutdown called for host "host", but we are on "(get-host-name))
	  (rmt:server-shutdown host portnum))
      ;; (bdat-time-to-exit-set! *bdat* #t) ;; tell on-exit to be fast as we've already cleaned up
      (portlogger:open-run-close portlogger:set-port port "released") ;; done in rmt:run
      ;; (debug:print-info 0 *default-log-port* "Max cached queries was    " *max-cache-size*)
      ;; (debug:print-info 0 *default-log-port* "Number of cached writes   " *number-of-writes*)


      ;; (debug:print-info 0 *default-log-port* "Average cached write time "
      ;; 		      (if (eq? *number-of-writes* 0)
      ;; 			  "n/a (no writes)"
      ;; 			  (/ *writes-total-delay*
      ;; 			     *number-of-writes*))

      ;; 		      " ms")
      ;; (debug:print-info 0 *default-log-port* "Number non-cached queries "  *number-non-write-queries*)
      ;; (debug:print-info 0 *default-log-port* "Average non-cached time   "
      ;; 		      (if (eq? *number-non-write-queries* 0)


      ;; 			  "n/a (no queries)"
      ;; 			  (/ *total-non-write-delay* 
      ;; 			     *number-non-write-queries*))
      ;; 		      " ms")
      
      (db:print-current-query-stats)
      (debug:print-info 0 *default-log-port* "Server shutdown complete. Exiting")
      )))

(define (rmt:try-start-server ipaddrstr portnum)
  (if *server-info* ;; update the server info as we might be trying next port
      (begin
	(servdat-host-set! *server-info* ipaddrstr)
	(servdat-port-set! *server-info* portnum)
	(servdat-status-set! *server-info* 'trying-port)
	(servdat-trynum-set! *server-info*
			     (+ (servdat-trynum *server-info*) 1)))
      (set! *server-info* (make-servdat host: ipaddrstr port: portnum)))
  (debug:print-info 0 *default-log-port* "rmt:try-start-server time="
		    (seconds->time-string (current-seconds))
		    " ipaddrsstr=" ipaddrstr
		    " portnum=" portnum)
  (assert (servdat? *server-info*) "FATAL: Must always have *server-info* properly set up by here.")
  (servdat-status-set! *server-info* 'starting)
  (servdat-port-set!   *server-info* portnum)
  (if (not (servdat-rep *server-info*))
      (let ((rep  (make-rep-socket)))
	(servdat-rep-set!    *server-info* rep)
	(socket-set! rep 'nng/recvtimeo 2000)))
  (let* ((rep (servdat-rep *server-info*)))
    (debug:print 0 *default-log-port* "INFO: Trying to start server on " ipaddrstr ":" portnum)
    (handle-exceptions
     exn
     (begin
       (print-error-message exn)
       (if (< portnum 64000)
	   (begin 
	     (debug:print 0 *default-log-port* "WARNING: attempt to start server failed. Trying again ...")
	     (debug:print 0 *default-log-port* " message: " ((condition-property-accessor 'exn 'message) exn))
	     (debug:print 5 *default-log-port* "exn=" (condition->list exn))
	     (portlogger:open-run-close portlogger:set-failed portnum)
	     (debug:print 0 *default-log-port* "WARNING: failed to start on portnum: " portnum ", trying next port")
	     ;; (thread-sleep! 0.1)
	     (rmt:try-start-server ipaddrstr
				   (portlogger:open-run-close portlogger:find-port)))
	   (begin
	     (print "ERROR: Tried and tried but could not start the server, stopping at port "portnum))))
     (nng-listen rep (conc "tcp://*:" portnum))
     rep)))

;;======================================================================
;; S E R V E R   U T I L I T I E S 
;;======================================================================

;;======================================================================
;; C L I E N T S

;; only use for main.db - need to re-write some of this :(
;;
(define (get-lock-db sdat dbfile host port)
  (assert host "FATAL: get-lock-db called with host not set.")
  (assert port "FATAL: get-lock-db called with port not set.")
  (let* ((dbh (db:open-run-db dbfile db:initialize-db)) ;; open-run-db creates a standard db with schema used by all situations
	 (res (db:get-iam-server-lock dbh dbfile host port)))

    ;; res => list then already locked, check server is responsive
    ;;     => #t then sucessfully got the lock
    ;;     => #f reserved for future use as to indicate something went wrong
    (match res
      ((owner_pid owner_host owner_port event_time)
       (if (server-ready? owner_host owner_port "abc")
	   #f      ;; locked by someone else
	   (begin  ;; locked by someone dead and gone
	     (debug:print 0 *default-log-port* "WARNING: stale lock - have to steal it. This may fail.")
	     (db:steal-lock-db dbh dbfile port))))
      (#t  #t) ;; placeholder so that we don't touch res if it is #t
      (else (set! res #f)))
    (sqlite3:finalize! dbh)

  (let* ((effective-toppath (or *toppath* apath)))
    (assert effective-toppath
	    "ERROR: get-pkts-dir called without *toppath* set. Exiting.")
    (let* ((pdir (conc effective-toppath "/.meta/srvpkts")))
      (if (file-exists? pdir)
	  pdir
	  (begin



	    (create-directory pdir #t)
	    pdir)))))

;; given a pkts dir read 
;;
(define (get-all-server-pkts pktsdir-in pktspec)
  (let* ((pktsdir  (if (file-exists? pktsdir-in)
		       pktsdir-in
		       (begin
			 (create-directory pktsdir-in #t)
			 pktsdir-in)))
	 (all-pkt-files (glob (conc pktsdir "/*.pkt"))))
    (map (lambda (pkt-file)
	   (read-pkt->alist pkt-file pktspec: pktspec))
	 all-pkt-files)))

(define (server-address srv-pkt)
  (conc (alist-ref 'host srv-pkt) ":"
	(alist-ref 'port srv-pkt)))
	
(define (server-ready? host port key) ;; server-address is host:port

  (let* ((data (sexpr->string  `((cmd . ping)
				 (key . ,key)
				 (params . ()))))
	 (res  (open-send-receive-nn (conc host ":" port) data)))

    (if res
	(string->sexpr res)
	res)))


; from the pkts return servers associated with dbpath
;; NOTE: Only one can be alive - have to check on each
;;       in the list of pkts returned
;;
(define (get-viable-servers serv-pkts dbpath)
  (let loop ((tail serv-pkts)
	     (res  '()))
    (if (null? tail)
	res ;; NOTE: sort by age so oldest is considered first
	(let* ((spkt (car tail)))
	  (loop (cdr tail)
		(if (equal? dbpath (alist-ref 'dbpath spkt))
		    (cons spkt res)
		    res))))))

(define (remove-pkts-if-not-alive serv-pkts)
  (filter (lambda (pkt)
	    (let* ((host (alist-ref 'host pkt))
		   (port (alist-ref 'port pkt))

		   (key  (alist-ref 'servkey  pkt))
		   (pktz (alist-ref 'Z        pkt))
		   (res  (server-ready? host port key)))
	      (if res
		  res
		  (let* ((pktsdir (get-pkts-dir *toppath*))
			 (pktpath (conc pktsdir"/"pktz".pkt")))
		    (debug:print 0 *default-log-port* "WARNING: pkt with no server "pktpath)
		    (delete-file* pktpath)
		    #f))))
	  serv-pkts))

;; from viable servers get one that is alive and ready
;;
(define (get-the-server apath serv-pkts)
  (let loop ((tail serv-pkts))
    (if (null? tail)
	#f
	(let* ((spkt  (car tail))
	       (host  (alist-ref 'ipaddr spkt))
	       (port  (alist-ref 'port spkt))

	       (dbpth (alist-ref 'dbpath spkt))
	       (srvkey (alist-ref 'Z spkt)) ;; (alist-ref 'srvkey spkt))
	       (addr  (server-address spkt)))
	  (if (server-ready? host port srvkey)
	      spkt
	      (loop (cdr tail)))))))

;; am I the "first" in line server? I.e. my D card is smallest
;; use Z card as tie breaker
;;
(define (get-best-candidate serv-pkts dbpath)

;;======================================================================
;; END NEW SERVER METHOD
;;======================================================================

;; if .db/main.db check the pkts
;; 
(define (rmt:wait-for-server pkts-dir db-file server-key)
  (let* ((sdat *server-info*))
    (let loop ((start-time (current-seconds))
	       (changed    #t)
	       (last-sdat  "not this"))
      (begin ;; let ((sdat #f))
	(thread-sleep! 0.01)
	(debug:print-info 0 *default-log-port* "Waiting for server alive signature")
	(mutex-lock! *heartbeat-mutex*)
	(set! sdat *server-info*)
	(mutex-unlock! *heartbeat-mutex*)
	(if (and sdat
		 (not changed)
		 (> (- (current-seconds) start-time) 2))
	    (begin
	      (servdat-status-set! sdat 'iface-stable)
	      (debug:print-info 0 *default-log-port* "Received server alive signature, now attempting to lock in server")
	      ;; create a server pkt in *toppath*/.meta/srvpkts
	      
	      ;; TODO:
	      ;;   1. change sdat to stuct
	      ;;   2. add uuid to struct
	      ;;   3. update uuid in sdat here
	      ;;
	      (servdat-uuid-set! sdat
				 (register-server
				  pkts-dir *srvpktspec*
				  (get-host-name)
				  (servdat-port sdat) server-key
				  (servdat-host sdat) db-file))
	      ;; (set! *my-signature* (servdat-uuid sdat)) ;; replace with Z, no, stick with proper key
	      ;; now read pkts and see if we are a contender
	      (let* ((all-pkts     (get-all-server-pkts pkts-dir *srvpktspec*))
		     (viables      (get-viable-servers all-pkts db-file))
		     (alive        (remove-pkts-if-not-alive viables))
		     (best-srv     (get-best-candidate alive db-file))
		     (best-srv-key (if best-srv (alist-ref 'servkey best-srv) #f))
		     (i-am-srv     (equal? best-srv-key server-key))
		     (delete-pkt   (lambda ()
				     (let* ((pktfile (conc (get-pkts-dir *toppath*)
							 "/" (servdat-uuid *server-info*)
							 ".pkt")))
				       (debug:print-info 0 *default-log-port* "Attempting to remove bogus pkt file "pktfile)
				       (delete-file* pktfile))))) ;; remove immediately instead of waiting for on-exit
		(debug:print 0 *default-log-port* "best-srv-key: "best-srv-key", server-key: "server-key", i-am-srv: "i-am-srv)
		;; am I the best-srv, compare server-keys to know
		(if i-am-srv
		    (if (get-lock-db sdat db-file (servdat-host sdat)(servdat-port sdat)) ;; (db:get-iam-server-lock *dbstruct-db* *toppath* run-id)

		  (begin
		    (debug:print-error 0 *default-log-port* "transport appears to have died, exiting server")
		    (exit))
		  (loop start-time
			(equal? sdat last-sdat)
			sdat))))))))

(define (rmt:register-server remdat apath iface port server-key dbname)
  (remotedat-conns remdat) ;; just checking types
  (rmt:open-main-connection remdat apath) ;; we need a channel to main.db
  (rmt:send-receive-real remdat apath      ;; params: host port servkey pid ipaddr dbpath
			 (db:run-id->dbname #f)
			 'register-server `(,iface
					    ,port
					    ,server-key
					    ,(current-process-id)
					    ,iface
					    ,apath
					    ,dbname)))

(define (rmt:get-count-servers remdat apath)
  (remotedat-conns remdat) ;; just checking types
  (rmt:open-main-connection remdat apath) ;; we need a channel to main.db
  (rmt:send-receive-real remdat apath      ;; params: host port servkey pid ipaddr dbpath
			 (db:run-id->dbname #f)
			 'get-count-servers `(,apath)))

(define (rmt:get-servers-info apath)
  (rmt:send-receive 'get-servers-info #f `(,apath)))

(define (rmt:deregister-server remdat apath iface port server-key dbname)
  (remotedat-conns remdat) ;; just checking types
  (rmt:open-main-connection remdat apath) ;; we need a channel to main.db
  (rmt:send-receive-real remdat apath      ;; params: host port servkey pid ipaddr dbpath
                         (db:run-id->dbname #f)
                         'deregister-server `(,iface
                                              ,port
                                              ,server-key
                                              ,(current-process-id)
                                              ,iface
                                              ,apath
                                              ,dbname)))

(define (rmt:wait-for-stable-interface #!optional (num-tries-allowed 100))
  ;; wait until *server-info* stops changing
  (let* ((stime (current-seconds)))
    (let loop ((last-host  #f)
	       (last-port  #f)
	       (tries 0))
      (let* ((curr-host (and *server-info* (servdat-host *server-info*)))
	     (curr-port (and *server-info* (servdat-port *server-info*))))
	;; first we verify port and interface, update *server-info* in need be.
	(cond
	 ((> tries num-tries-allowed)
	  (debug:print 0 *default-log-port* "rmt:keep-running, giving up after trying for several minutes.")
	  (exit 1))
	 ((not *server-info*)
	  (thread-sleep! 0.25)
	  (loop curr-host curr-port (+ tries 1)))
	 ((or (not last-host)(not last-port))
	  (debug:print 0 *default-log-port* "rmt:keep-running, still no interface, tries="tries)
	  (thread-sleep! 0.25)
	  (loop curr-host curr-port (+ tries 1)))
	 ((or (not (equal? last-host curr-host))
	      (not (equal? last-port curr-port)))
	  (debug:print-info 0 *default-log-port* "WARNING: interface changed, refreshing iface and port info")
	  (thread-sleep! 0.25)
	  (loop curr-host curr-port (+ tries 1)))
	 ((< (- (current-seconds) stime) 1) ;; keep up the looping until at least 3 seconds have passed
	  (thread-sleep! 0.5)
	  (loop curr-host curr-port (+ tries 1)))
	 (else
	  (rmt:get-signature) ;; sets *my-signature* as side effect
	  (servdat-status-set! *server-info* 'interface-stable)
	  (debug:print 0 *default-log-port*
		       "SERVER STARTED: " curr-host
		       ":" curr-port
		       " AT " (current-seconds) " server signature: " *my-signature*
		       " with "(servdat-trynum *server-info*)" port changes")
	  (flush-output *default-log-port*)
	  #t))))))

;; run rmt:keep-running in a parallel thread to monitor that the db is being 
;; used and to shutdown after sometime if it is not.
;;
(define (rmt:keep-running dbname) 
  ;; if none running or if > 20 seconds since 
  ;; server last used then start shutdown
  ;; This thread waits for the server to come alive
  (debug:print-info 0 *default-log-port* "Starting the sync-back, keep alive thread in server")

  (let* ((remdat            *remotedat*)
	 (server-start-time (current-seconds))
	 (pkts-dir          (get-pkts-dir))
	 (server-key        (rmt:get-signature)) ;; This servers key
	 (is-main           (equal? (args:get-arg "-db") ".db/main.db"))
	 (last-access       0)
	 (server-timeout    (server:expiration-timeout))
	 (shutdown-server-sequence (lambda (host port)
				     (set! *unclean-shutdown* #f)
				     (debug:print-info 0 *default-log-port* "Starting to shutdown the server. pid="(current-process-id))
				     (rmt:server-shutdown host port)
				     (portlogger:open-run-close portlogger:set-port port "released")
				     (exit)))
	 (timed-out?        (lambda ()
			      (<= (+ last-access server-timeout)
				  (current-seconds)))))
    (servdat-dbfile-set! *server-info* (args:get-arg "-db"))
    ;; main and run db servers have both got wait logic (could/should merge it)
    (if is-main
	(rmt:wait-for-server pkts-dir dbname server-key)
	(rmt:wait-for-stable-interface))
    ;; this is our forever loop
    (let* ((iface             (servdat-host *server-info*))
	   (port              (servdat-port *server-info*)))

      (let loop ((count          0)
		 (bad-sync-count 0)
		 (start-time     (current-milliseconds)))
	(if (and (not is-main)
		 (common:low-noise-print 60 "servdat-status"))
	    (debug:print-info 0 *default-log-port* "servdat-status is " (servdat-status *server-info*)))

	;; set up the database handle
	(mutex-lock! *heartbeat-mutex*)

	(if (not *dbstruct-db*) ;; no db opened yet, open the db and register with main if appropriate
	    (let ((watchdog (bdat-watchdog *bdat*)))		 
	      (debug:print 0 *default-log-port* "SERVER: dbprep")
	      (db:setup dbname) ;; sets *dbstruct-db* as side effect
	      (servdat-status-set! *server-info* 'db-opened)
	      ;; IFF I'm not main, call into main and register self
	      (if (not is-main)
		  (let ((res (rmt:register-server remdat
						  *toppath* iface port
						  server-key dbname)))
		    (if res ;; we are the server
			(servdat-status-set! *server-info* 'have-interface-and-db)
			;; now check that the db locker is alive, clear it out if not
			(let* ((serv-info (rmt:get-server-info *toppath* dbname)))
			  (match serv-info
			    ((host port servkey pid ipaddr apath dbpath)
			     (if (not (server-ready? host port servkey))
				 (begin
				   (debug:print-info 0 *default-log-port* "Server registered but not alive. Removing and trying again.")
				   (rmt:deregister-server remdat apath host port servkey dbpath) ;; servkey pid ipaddr apath dbpath)
				   
				   (loop (+ count 1) bad-sync-count start-time))))
			    (else
			     (debug:print 0 *default-log-port* "We are not the server for "dbname", exiting. Server info is: "serv-info)
			     (exit)))))))
	      (debug:print 0 *default-log-port*
			   "SERVER: running, db "dbname" opened, megatest version: "
			   (common:get-full-version))
	      ;; start the watchdog

	      ;; is this really needed?
	      
	      #;(if watchdog
		  (if (not (member (thread-state watchdog)
				   '(ready running blocked
					   sleeping dead)))
		      (begin
			(debug:print-info 0 *default-log-port* "Starting watchdog thread (in state "(thread-state watchdog)")")
			(thread-start! watchdog))
		      (debug:print-info 0 *default-log-port* "Not starting watchdog thread (in state "(thread-state watchdog)")"))
		  (debug:print 0 *default-log-port* "ERROR: *watchdog* not setup, cannot start it."))
	      #;(loop (+ count 1) bad-sync-count start-time)
	      ))

	(debug:print-info 0 *default-log-port* "syncing "*toppath*" "dbname" at "(current-seconds))
	(db:sync-inmem->disk *dbstruct-db* *toppath* dbname force-sync: #t)
	
	(mutex-unlock! *heartbeat-mutex*)
	
	;; when things go wrong we don't want to be doing the various
	;; queries too often so we strive to run this stuff only every
	;; four seconds or so.
	(let* ((sync-time (- (current-milliseconds) start-time))
	       (rem-time  (quotient (- 4000 sync-time) 1000)))
	  (if (and (<= rem-time 4)
		   (>  rem-time 0))
	      (thread-sleep! rem-time)))



	
	(if (< count 1) ;; 3x3 = 9 secs aprox
	    (loop (+ count 1) bad-sync-count (current-milliseconds)))
	
	;; Transfer *db-last-access* to last-access to use in checking that we are still alive
	(set! last-access *db-last-access*)
	
	(if (common:low-noise-print 60 "dbstats")
	    (begin
	      (debug:print 0 *default-log-port* "Server stats:")
	      (db:print-current-query-stats)))
	(let* ((hrs-since-start  (/ (- (current-seconds) server-start-time) 3600)))
	  (cond
	   ((not *server-run*)
	    (debug:print-info 0 *default-log-port* "*server-run* set to #f. Shutting down.")
	    (shutdown-server-sequence (get-host-name) port))
	   ((timed-out?)
	    (debug:print-info 0 *default-log-port* "Server timed out. seconds since last db access: " (- (current-seconds) last-access))
	    (shutdown-server-sequence (get-host-name) port))
	   ((and *server-run*
		 (or (not (timed-out?))
		     (if is-main ;; do not exit if there are other servers (keep main open until all others gone)
			 (> (rmt:get-count-servers remdat *toppath*) 1)
			 #f)))
	    (if (common:low-noise-print 120 "server continuing")
		(debug:print-info 0 *default-log-port* "Server continuing, seconds since last db access: " (- (current-seconds) last-access)))
	    (loop 0 bad-sync-count (current-milliseconds)))
	   (else
	    (set! *unclean-shutdown* #f)
	    (debug:print-info 0 *default-log-port* "Server timed out. seconds since last db access: " (- (current-seconds) last-access))
	    (shutdown-server-sequence (get-host-name) port)
	    #;(debug:print-info 0 *default-log-port* "Sending 'quit to server, received: "
			      (open-send-receive-nn (conc iface":"port)      ;; do this here and not in server-shutdown
						    (sexpr->string 'quit)))





	    )))))))

;; Call this to start the actual server
;;
;; all routes though here end in exit ...
;;
;; This is the point at which servers are started
;;
(define (rmt:server-launch dbname)
  (debug:print-info 0 *default-log-port* "Entered rmt:server-launch")
  (let* ((th2 (make-thread (lambda ()
			     (debug:print-info 0 *default-log-port* "Server run thread started")
			     (rmt:run (if (args:get-arg "-server")
					  (args:get-arg "-server")
					  "-")
				      )) "Server run"))
	 (th3 (make-thread (lambda ()
			     (debug:print-info 0 *default-log-port* "Server monitor thread started")

			     (rmt:keep-running dbname)
			     "Keep running"))))
    (thread-start! th2)
    (thread-sleep! 0.252) ;; give the server time to settle before starting the keep-running monitor.
    (thread-start! th3)
    (set! *didsomething* #t)
    (thread-join! th2)
    (thread-join! th3)
    ;; (exit))
  )
  #f
  )
	    
;; Generate a unique signature for this process, used at both client and
;; server side
(define (rmt:mk-signature)
  (message-digest-string (md5-primitive) 
			 (with-output-to-string
			   (lambda ()

	      (if (string-search (regexp (conc ":" port-num)) inl)
		  (begin
					;(print "Output: "  inl)
		    (set! ret  #t))
		  (loop (read-line inp)))))))
    ret))

;;start a server, returns the connection
;;
(define (start-nn-server portnum )
  (let ((rep (make-rep-socket))) ;; (nn-socket 'rep)))
    (socket-set! rep 'nng/recvtimeo 2000)
    (handle-exceptions ;; why have exception handler here? 
     exn
     (let ((emsg ((condition-property-accessor 'exn 'message) exn)))
       (print "ERROR: Failed to start server \"" emsg "\"")
       (exit 1))
      
     (nng-dial #;nn-bind rep (conc "tcp://*:" portnum)))
    rep))

(define (open-nn-connection host-port)
  (let ((req  (make-req-socket))
        (uri  (conc "tcp://" host-port)))
    (nng-dial req uri)
    (socket-set! req 'nng/recvtimeo 2000)
    req))

(define (send-receive-nn req msg)
  (nng-send req msg)
  (nng-recv req))

(define (close-nn-connection req)
  (nng-close! req))
  
;; ;; open connection to server, send message, close connection
;; ;;
;; (define (open-send-close-nn host-port msg #!key (timeout 3) ) ;; default timeout is 3 seconds
;;   (let ((req  (make-req-socket 'req))
;;         (uri  (conc "tcp://" host-port))

;;                                 (thread-terminate! th1))
;; 			      "timer thread")))
;;        (thread-start! th1)
;;        (thread-start! th2)
;;        (thread-join! th1)
;;        res))))
;; 
(define (open-send-receive-nn host-port msg #!key (timeout 3) ) ;; default timeout is 3 seconds
  (let ((req  (make-req-socket))
        (uri  (conc "tcp://" host-port))
        (res  #f)) 
    (handle-exceptions
     exn
     (let ((emsg ((condition-property-accessor 'exn 'message) exn)))
       ;; Send notification