Megatest

(module dbmod
	*
	
(import scheme
	chicken
	data-structures
	extras
	files

	(prefix sqlite3 sqlite3:)
	matchable
	posix
	typed-records
	srfi-1
	srfi-18
	srfi-69

	commonmod

	   (curr-secs   (current-seconds)))
      (if (> (- curr-secs last-update) 3)
	  (begin
	    (sync-proc last-update)

	    ;; MOVE THIS CALL TO INSIDE THE sync-proc CALL
	    (dbr:dbstruct-last-update-set! dbstruct curr-secs)

	    )))
    (assert (sqlite3:database? dbh) "FATAL: bad db handle in dbmod:with-db") 
    (if use-mutex (mutex-lock! *db-with-db-mutex*))
    (let* ((res (apply proc dbdat dbh params)))
      (if use-mutex (mutex-unlock! *db-with-db-mutex*))
      res)))

   (lambda ()
     (let* ((db      (sqlite3:open-database dbfullname))
	    (handler (sqlite3:make-busy-timeout 136000)))
       (sqlite3:set-busy-handler! db handler)
       (if write-access
	   (init-proc db))
       db))))

;; try every second until tries times proc
;;
(define (db:keep-trying-until-true proc params tries)
  (let* ((res (apply proc params)))
    (if res
	res
	(if (> tries 0)
	    (begin
	      (thread-sleep! 1)
	      (db:keep-trying-until-true proc params (- tries 1)))
	    (begin
	      ;; (debug:print-info 0 *default-log-port* "proc never returned true, params="params)
	      (print"db:keep-trying-until-true proc never returned true, proc = " proc " params =" params " tries = " tries)
	      #f)))))
  

(define *sync-in-progress* #f)

;; Open the inmem db and the on-disk db
;; populate the inmem db with data
;;
;; Updates fields in dbstruct

	 (dbexists     (file-exists? dbfullname))
	 (tmpdir       (conc "/tmp/"(current-user-name)))
	 (tmpdb        (let* ((fname (conc tmpdir"/" (string-translate areapath "/" ".")"-"(current-process-id)"-"dbfname)))
			 (if (not (file-exists? tmpdir))(create-directory tmpdir))
			 ;; check if tmpdb already exists, either delete it or
			 ;; add something to the name
			 fname))
	 #;(inmem        (dbmod:open-inmem-db init-proc
					    (if (eq? (dbcache-mode) 'inmem)
						#f
						tmpdb)
					    ))
	 (write-access (file-write-access? dbpath))
	 (db           (dbmod:safely-open-db dbfullname init-proc write-access))
	 (tables       (db:sync-all-tables-list keys)))

    (if (not (sqlite3:database? db)) ;; db is our master database in the .mtdb dir
	(begin
	  (debug:print 0 *default-log-port* "ERROR: Failed to properly open "dbfname-in", exiting immediately.")
	  (exit)))
    ;; we sync to tmpdb here so that we use file-copy to get intial database
    (dbmod:db-to-db-sync dbfullname tmpdb 0 init-proc keys)
    (let* ((inmem (dbmod:open-inmem-db init-proc tmpdb)))
      (dbr:dbstruct-inmem-set!     dbstruct inmem))
    (dbr:dbstruct-ondiskdb-set!  dbstruct db)
    (dbr:dbstruct-dbfile-set!    dbstruct dbfullname)
;    (dbr:dbstruct-dbfname-set!   dbstruct dbfname)
    (dbr:dbstruct-sync-proc-set! dbstruct
				 (lambda (last-update)
				   (if *sync-in-progress*
				       (debug:print 3 *default-log-port* "WARNING: overlapping calls to sync to disk")
				       (thread-start!
					(make-thread
					 (lambda ()
					   (mutex-lock! *db-with-db-mutex*) ;; this mutex is used when overloaded or during a query that modifies the db
					   (set! *sync-in-progress* #t)
					   #;(dbmod:sync-gasket tables last-update inmem db
					   dbfullname syncdir)
					   (system (conc "megatest -db2db -from "tmpdb" -to "dbfullname))
					   (mutex-unlock! *db-with-db-mutex*)
					   (thread-sleep! 0.5) ;; ensure at least 1/2 second down time between sync calls
					   (set! *sync-in-progress* #f)))))))



    (dbr:dbstruct-last-update-set! dbstruct (current-seconds)) ;; should this be offset back in time by one second?
    dbstruct))

;;    (if (eq? syncdir 'todisk) ;; sync to disk normally, sync from in dashboard
;;        (dbmod:sync-tables tables last-update inmem db)
;;        (dbmod:sync-tables tables last-update db inmem))))

;; direction: 'fromdest 'todest
;;
(define (dbmod:sync-gasket tables last-update inmem dbh dbfname direction keys)
  (assert (sqlite3:database? inmem) "FATAL: sync-gasket: inmem is not a db")
  (assert (sqlite3:database? dbh) "FATAL: sync-gasket: dbh is not a db")
  (debug:print-info 0 *default-log-port* "Db sync using "(dbfile:sync-method)" method")
  (case (dbfile:sync-method)
    ((none) #f)
    ((attach)
     (dbmod:attach-sync tables inmem dbfname direction))
    ((newsync) ;; DON'T USE THIS ONE. IT IS BORKED
     (dbmod:new-sync tables inmem dbh dbfname direction))
    (else
     (case direction
       ((todisk)
	(dbmod:sync-tables tables last-update keys inmem dbh)
	)
       (else
	(dbmod:sync-tables tables last-update keys dbh inmem))))))

(define (dbmod:close-db dbstruct)
  ;; do final sync to disk file
  ;; (do-sync ...)
  (sqlite3:finalize! (dbr:dbstruct-ondiskdb dbstruct)))

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

;;
;; if last-update specified ("field-name" . time-in-seconds)
;;    then sync only records where field-name >= time-in-seconds
;;    IFF field-name exists
;;
;; Use (db:sync-all-tables-list keys) to get the tbls input
;;
(define (dbmod:sync-tables tbls last-update keys fromdb todb)
  (debug:print-info 0 *default-log-port* "dbmod:sync-tables called, from: "fromdb", to: "todb)
  (assert (sqlite3:database? fromdb) "FATAL: dbmod:sync-tables called with fromdb not a database" fromdb)
  (assert (sqlite3:database? todb) "FATAL: dbmod:sync-tables called with fromdb not a database" todb)
  (let ((specials    `(("keys" "fieldname")
		       ("metadat" "var")
		       ,(cons "runs" (cons "runname" keys))
		       ("tests" "run_id" "testname" "item_path")
		       ("test_meta" "testname")
		       ("test_steps" "test_id" "stepname" "state")
		       ("test_data" "test_id" "category" "variable")))
	(stmts       (make-hash-table)) ;; table-field => stmt
	(all-stmts   '())              ;; ( ( stmt1 value1 ) ( stml2 value2 ))
	(numrecs     (make-hash-table))
	(start-time  (current-milliseconds))
	(tot-count   0))
    (for-each ;; table
     (lambda (tabledat)





























       (let* ((count (match tabledat







		       ((tablename . fields)
			(debug:print-info 0 *default-log-port* "Syncing table "tablename)













			(dbmod:sync-table tablename fields fromdb todb specials))



		       (else



			(debug:print-warn 0 *default-log-port* "Bad tabledat entry: "tabledat)











			0))))
































	 (set! tot-count (+ tot-count count))))




     tbls)
    (debug:print-info 0 *default-log-port* "dbmod:sync-tables completed in "(- (current-milliseconds) start-time)"ms")









    tot-count))

(define (dbmod:sync-table tablename fields from-db to-db specials)
  (let* ((key-fields       (alist-ref tablename specials equal?))
	 (field-names      (map car fields))
	 (has-last-update  (member "last_update" field-names))
	 (fields-sans-lu   (filter (lambda (x)
				     (not (member x '("id" "last_update"))))
				   field-names))
	 (get-vals        (lambda (db id fields)
			    (debug:print-info 0 *default-log-port* "get-vals: fields="fields", id="id)
			    (let* ((qry (conc "SELECT "(string-intersperse fields ",")" FROM "tablename" WHERE id=?;"))
				   (res #f))
			      (sqlite3:for-each-row
			       (lambda tuple
				 (set! res tuple))
			       db qry id)
			      res)))
	 (clean-up-qry    (lambda (from-id)
			    (debug:print-info 0 *default-log-port* "key-fields="key-fields", from-id="from-id)
			    (let* ((vals (get-vals from-db from-id key-fields))
				   (qry  (conc "DELETE FROM "tablename" WHERE "(string-intersperse key-fields "=? AND ")"=?;")))
			      (debug:print-info 0 *default-log-port* "qry: "qry", vals="vals)
			      (apply sqlite3:execute to-db qry vals))))
	 (get-ids          (lambda (db)
			     (sqlite3:fold-row (lambda (res id)
						 (cons id res))
					       '()
					       db
					       (conc "SELECT id FROM "tablename";"))))
	 (get-val          (lambda (db fieldname id)
			     (let* ((res #f)
				    (sql (conc "SELECT "fieldname" FROM "tablename" WHERE id=?;")))
			       (sqlite3:for-each-row
				(lambda (val)
				  (set! res val))
				db
				sql
				id)
			       ;; (debug:print-info 0 *default-log-port* "get-val "db" "fieldname" "id", sql="sql", res="res)
			       res)))
	 (get-row         (lambda (db id)
			    (let* ((res #f))
			      (sqlite3:for-each-row
			       (lambda tuple
				 (set! res tuple))
			       db
			       (conc "SELECT " (string-intersperse fields-sans-lu ",")
				     " FROM "tablename" WHERE id=?;")
			       id)
			      res)))
	 (ins-row         (lambda (db id row)
			    (let* ((qry (conc "INSERT INTO "tablename" (id,"
					      (string-intersperse fields-sans-lu ",")
					      ") VALUES ("id","
					      (string-intersperse
					       (make-list (length fields-sans-lu) "?")
					       ",")
					      ");"))
				   (proc (lambda ()
					   (apply sqlite3:execute db qry row))))
			      ;; (debug:print-info 0 *default-log-port* "qry="qry)
			      (handle-exceptions ;; on exception do the cleanup qry then try one more time
				  exn
				(begin
				  ;; (clean-up-qry id)
				  (proc))
				(proc)))))
			      
	 (num-inserts     0)
	 (num-updates     0)
	 )
    ;; (debug:print-info 0 *default-log-port* "field-names: "field-names", fields-sans-lu: "fields-sans-lu)
    ;; (sqlite3:with-transaction
    ;;  from-db
    ;;  (lambda ()
       (let* ((from-ids (get-ids from-db)))
	 ;; (debug:print-info 0 *default-log-port* "Table "tablename", has "(length from-ids)" records.")
 	 (sqlite3:with-transaction
 	  to-db
	  (lambda ()
	    (let* ((to-ids (get-ids to-db)))
	      ;; (debug:print 0 *default-log-port* "to-ids="to-ids)
	      (for-each ;; from-id
	       (lambda (from-id)
		 (if (member from-id to-ids)
		     (for-each ;; case where record exists, do one by one the fields if different
		      (lambda (fieldname)
			(let* ((from-val (get-val from-db fieldname from-id))
			       (dest-val (get-val to-db   fieldname from-id)))
			  #;(debug:print 0 *default-log-port*
				       "fieldname="fieldname
				       ", from-id="from-id
				       ", from-val="from-val
				       ", dest-val="dest-val
				       )
			  (if (not (equal? from-val dest-val))
			      (let* ((qry-proc (lambda ()
						 (sqlite3:execute to-db (conc "UPDATE "tablename" SET "fieldname"=? WHERE id=?;")
								  from-val from-id))))
				(handle-exceptions ;; try to remove the offending record and re-try once the update
				    exn
				  (begin
				    ;; (clean-up-qry from-id)
				    (qry-proc))
				  (qry-proc))
				(set! num-updates (+ num-updates 1))))))
		      fields-sans-lu)
		     (let ((row (get-row from-db from-id))) ;; need to insert the row
		       (debug:print 0 *default-log-port* "from-id="from-id", to-ids="to-ids", row="row)
		       (set! num-inserts (+ num-inserts 1))
		       (ins-row to-db from-id row))))
	       from-ids)))))
    (+ num-inserts num-updates)))

;;     (for-each ;; table
;;      (lambda (tabledat)
;;        (let* ((tablename        (car tabledat))
;; 	      (fields           (cdr tabledat))
;; 	      (has-last-update  (member "last_update" fields))
;; 	      (use-last-update  (dbmod:calc-use-last-update has-last-update fields last-update))
;; 	      (last-update-value (if use-last-update ;; no need to check for has-last-update - it is already accounted for
;; 				     (if (number? last-update)
;; 					 last-update
;; 					 (cdr last-update))
;; 				     #f))
;; 	      (last-update-field (if use-last-update
;; 				     (if (number? last-update)
;; 					 "last_update"
;; 					 (car last-update))
;; 				     #f))
;; 	      (num-fields (length fields))
;; 	      (field->num (make-hash-table))
;; 	      (num->field (apply vector (map car fields))) ;; BBHERE
;; 	      (full-sel   (conc "SELECT " (string-intersperse (map car fields) ",") 
;; 				" FROM " tablename (if use-last-update ;; apply last-update criteria
;; 						       (conc " WHERE " last-update-field " >= " last-update-value)
;; 						       "")
;; 				";"))
;; 	      (full-ins   (conc "INSERT OR REPLACE INTO " tablename " ( " (string-intersperse (map car fields) ",") " ) "
;; 				" VALUES ( " (string-intersperse (make-list num-fields "?") ",") " );"))
;; 	      (fromdat    '())
;; 	      (fromdats   '())
;; 	      (totrecords 0)
;; 	      (batch-len  10000000) ;; (string->number (or (configf:lookup *configdat* "sync" "batchsize") "100")))
;; 	      (todat      (make-hash-table))
;; 	      (count      0)
;;               (field-names (map car fields)))
;; 	 
;; 	 (debug:print-info 0 *default-log-port* "Syncing table "tablename)
;; 	 
;; 	 ;; set up the field->num table
;; 	 (for-each
;; 	  (lambda (field)
;; 	    (hash-table-set! field->num field count)
;; 	    (set! count (+ count 1)))
;; 	  fields)
;; 	 
;; 	 ;; read the source table
;;          ;; store a list of all rows in the table in fromdat, up to batch-len.
;;          ;; Then add fromdat to the fromdats list, clear fromdat and repeat.
;; 	 (sqlite3:for-each-row
;; 	  (lambda (a . b)
;; 	       (set! fromdat (cons (apply vector a b) fromdat))
;; 	       (if (> (length fromdat) batch-len)
;; 		   (begin
;; 		     (set! fromdats (cons fromdat fromdats))
;; 		     (set! fromdat  '())
;; 		     (set! totrecords (+ totrecords 1)))))
;; 	  fromdb
;; 	  full-sel)
;; 
;; 	 (debug:print-info 0 *default-log-port* "Have "totrecords" records to update.")
;;          ;; Count less than batch-len as a record
;;          (if (> (length fromdat) 0)
;;              (set! totrecords (+ totrecords 1)))
;; 	 
;; 	 ;; tack on remaining records in fromdat
;; 	 (if (not (null? fromdat))
;; 	     (set! fromdats (cons fromdat fromdats)))
;; 	 
;; 	 (sqlite3:for-each-row
;; 	  (lambda (a . b)
;; 	    (hash-table-set! todat a (apply vector a b)))
;; 	  todb
;; 	  full-sel)
;; 	 
;; 	 ;; first pass implementation, just insert all changed rows
;; 	 
;; 	 (let* ((db                 todb)
;; 		(has-last-update    (member "last_update" field-names))
;;                 (drp-trigger        (if has-last-update
;; 					(db:drop-trigger db tablename) 
;; 					#f))
;;                 (is-trigger-dropped (if has-last-update
;;                                         (db:is-trigger-dropped db tablename)
;; 					#f))
;; 		(stmth              (sqlite3:prepare db full-ins))
;;                 (changed-rows       0))
;; 	   (for-each
;; 	    (lambda (fromdat-lst)
;; 	      (mutex-lock! *db-transaction-mutex*)
;; 	      (sqlite3:with-transaction
;; 	       db
;; 	       (lambda ()
;; 		 (for-each ;; 
;; 		  (lambda (fromrow)
;; 		    (let* ((a    (vector-ref fromrow 0))
;; 			   (curr (hash-table-ref/default todat a #f))
;; 			   (same #t))
;; 		      (let loop ((i 0))
;; 			(if (or (not curr)
;; 				(not (equal? (vector-ref fromrow i)(vector-ref curr i))))
;; 			    (set! same #f))
;; 			(if (and same
;; 				 (< i (- num-fields 1)))
;; 			    (loop (+ i 1))))
;; 		      (if (not same)
;; 			  (begin
;; 			    (apply sqlite3:execute stmth (vector->list fromrow))
;; 			    (hash-table-set! numrecs tablename (+ 1 (hash-table-ref/default numrecs tablename 0)))
;; 			    (set! changed-rows (+ changed-rows 1))))))
;; 		  fromdat-lst)))
;; 	      (mutex-unlock! *db-transaction-mutex*))
;; 	    fromdats)
;; 	   
;; 	   (sqlite3:finalize! stmth)
;;            (if (member "last_update" field-names)
;;                (db:create-trigger db tablename)))
;; 	 ))
;;      tbls)
;;     (let* ((runtime      (- (current-milliseconds) start-time))
;; 	   (should-print (or ;; (debug:debug-mode 12)
;; 			  (common:low-noise-print 120 "db sync")
;; 			  (> runtime 500)))) ;; low and high sync times treated as separate.
;;       (for-each 
;;        (lambda (dat)
;; 	 (let ((tblname (car dat))
;; 	       (count   (cdr dat)))
;; 	   (set! tot-count (+ tot-count count)))) 
;;        (sort (hash-table->alist numrecs)(lambda (a b)(> (cdr a)(cdr b))))))

(define (has-last-update dbh tablename)
  (let* ((has-last #f))
    (sqlite3:for-each-row
     (lambda (name)
       (if (equal? name "last_update")
	   (set! has-last #t)))
     dbh

			   (mode 'full)
			   (no-update '("keys")) ;; do
			   )
  (debug:print 0 *default-log-port* "Doing sync "direction" "destdbfile)
  (if (not (sqlite3:auto-committing? dbh))
      (debug:print 0 *default-log-port* "Skipping sync due to transaction in flight.")
      (let* ((table-names  (map car tables))
	     (dest-exists  (file-exists? destdbfile))
	     (start-time   (current-milliseconds)))
	(assert dest-exists "FATAL: sync called with non-existant file, "destdbfile)
	;; attach the destdbfile
	;; for each table
	;;    insert into dest.<table> select * from src.<table> where last_update>last_update
	;; done
	(debug:print 0 *default-log-port* "Attaching "destdbfile" as auxdb")
	(sqlite3:execute dbh (conc "ATTACH '"destdbfile"' AS auxdb;"))

	     ;; (debug:print 0 *default-log-port* "stmt8="stmt8)
	     ;; (if (sqlite3:auto-committing? dbh)
	     ;; (begin
	     (mutex-lock! *db-transaction-mutex*)
	     (sqlite3:with-transaction
	      dbh
	      (lambda ()
		(debug:print-info 0 *default-log-port* "Sync from "fromdb table" to "todb table" using INSERT OR UPDATE")
		(sqlite3:execute dbh stmt1)    ;; get all new rows
		
		(if (member "last_update" fields)
		    (sqlite3:execute dbh stmt8))    ;; get all updated rows
		;; (sqlite3:execute dbh stmt5)
		;; (sqlite3:execute dbh stmt4) ;; if it worked this would be better for incremental up
		;; (sqlite3:execute dbh stmt6)
		))
	     (debug:print 0 *default-log-port* "Synced table "table
			  " in "(- (current-milliseconds) start-ms)"ms") ;; )
	     (mutex-unlock! *db-transaction-mutex*)))
	     
	 ;; (debug:print 0 *default-log-port* "Skipping sync of table "table" due to transaction in flight."))))
	 table-names)
	(sqlite3:execute dbh "DETACH auxdb;")
	(debug:print-info 0 *default-log-port* "Total sync time: "(- (current-milliseconds) start-time)"ms")
	-1)))

;; FAILED ATTEMPTS

	     ;; (if (not (has-last-update dbh table))
	     ;;     (sqlite3:execute dbh (conc "ALTER TABLE "table" ADD COLUMN last_update INTEGER;")))
	     ;; (if (not (has-last-update dbh (conc "auxdb."table)))
	     ;;     (sqlite3:execute dbh (conc "ALTER TABLE auxdb."table" ADD COLUMN last_update INTEGER;")))

		(begin
		  (debug:print-info 0 *default-log-port* "Found old test in LAUNCHED state, test-id=" test-id
				    " 1 day since event_time marked")
                  (set! oldlaunched (cons (list test-id run-dir uname testname item-path run-id) oldlaunched)))))
	  stmth3
	  run-id))))
    (list incompleted oldlaunched toplevels)))

(define (db:set-state-status-by-state-status dbstruct run-id testname currstate currstatus newstate newstatus)


  ;; clear caches needed

  
  (let* ((qry (conc "UPDATE tests SET state=?,status=? WHERE "
		    (if currstate  (conc "state='" currstate "' AND ") "")
		    (if currstatus (conc "status='" currstatus "' AND ") "")
		    " run_id=? AND testname LIKE ?;")))
    (db:with-db
     dbstruct
     run-id
     #t
     (lambda (dbdat db)
       (sqlite3:execute db qry
			(or newstate  currstate "NOT_STARTED")
			(or newstatus currstate "UNKNOWN")
			run-id testname)))))
;;======================================================================
;; db to db sync
;;======================================================================

(define (dbmod:db-to-db-sync src-db dest-db last-update init-proc keys)
  (if (and (file-exists? src-db) ;; can't proceed without a source
	   (file-read-access? src-db))
      (let* ((have-dest     (file-exists? dest-db))
	     (dest-file-wr  (and have-dest
				 (file-write-access? dest-db))) ;; exists and writable
	     (dest-dir      (or (pathname-directory dest-db)
				"."))
	     (dest-dir-wr   (and (file-exists? dest-dir)
				 (file-write-access? dest-dir)))
	     (d-wr          (or (and have-dest
				     dest-file-wr)
				dest-dir-wr))
	     (copied        (if (and (not have-dest)
				     dest-dir-wr)
				(begin
				  (file-copy src-db dest-db)
				  #t)
				#f)))
	(if copied
	    (begin
	      (debug:print-info 0 *default-log-port* "db-to-db-sync done with file-copy")
	      #t)
	    (let* ((tables (db:sync-all-tables-list keys))
		   (sdb    (dbmod:safely-open-db src-db init-proc #t))
		   (ddb    (dbmod:safely-open-db dest-db init-proc d-wr))
		   (res    (dbmod:sync-gasket tables last-update sdb ddb dest-db 'todest keys)))
	      (sqlite3:finalize! sdb)
	      (sqlite3:finalize! ddb)
	      res)))
      #f))

)