--[[----------------------------------------------------------------------------

WFSQLiteDatabase.lua

--------------------------------------------------------------------------------

ADOBE CONFIDENTIAL
------------------
Copyright 2011 Adobe Systems Incorporated
All Rights Reserved.

NOTICE: All information contained herein is, and remain the property of
Adobe Systems Incorporated and its suppliers, if any. The intellectual and
technical concepts contained herein are proprietary to Adobe Systems
Incorporated and its suppliers and may be covered by U.S. and Foreign
Patents, patents in process, and are protected by trade secret or copyright
law.  Dissemination of this information or reproduction of this material is
strictly forbidden unless prior written permission is obtained from
Adobe Systems Incorporated.

------------------------------------------------------------------------------]]

	-- RxLua
local Rx = require 'Rx'
local Observable = Rx.Observable

	-- WF.core.diagnostic_log
local WFDiagnosticLog = require 'WFDiagnosticLog'

	-- WF.core.logging
local WFLogger = require 'WFLogger'

	-- WF.core.lua.threading
local WFCurrentThread = require 'WFCurrentThread'

	-- WF.core.string
local KSStringUtils = require 'KSStringUtils'

	-- WF.core.os
local KSDate = require 'KSDate'
local WFFileUtils = require 'WFFileUtils'

	-- WF.core.rxlua
local WFRxScheduler = require 'WFRxScheduler'

	-- WF.core.utils
local WFArguments = require 'WFArguments'
local WFCloseSupport = require 'WFCloseSupport'

	-- WF.sqlite.database
local WFSQLiteDatabaseImp = require 'WFSQLiteDatabaseImp'
local WFSQLiteCorruptionLogger = require 'WFSQLiteCorruptionLogger'


--------------------------------------------------------------------------------

local COLLECT_DIAGNOSTIC_LOG_DATA = true

local check = assert( WFArguments.check )

local logger = WFLogger 'WFSQLiteDatabase'

local useSharedLogger = 
		require "WFConfiguration".WFSQLiteDatabase.useSharedLogger

local disableExclusiveLocking = 
		require "WFConfiguration".WFSQLiteDatabase.disableExclusiveLocking

--------------------------------------------------------------------------------

-- SQLite Error Codes. Pulled from sqlite3.h

local SQLITE_IOERR_ERROR 	= 10 	-- Some kind of disk I/O error occurred
local SQLITE_CORRUPT_ERROR 	= 11 	-- The database disk image is malformed
local SQLITE_FULL_ERROR 	= 13 	-- Insertion failed because database is full

local initializationErrors = {
	standardError = "STANDARD_ERROR",
	conflictingSchemaDefinitionError = "CONFLICTING_SCHEMA"
}

local kDocStoreInitProgress = 'docStoreInitProgress'

--============================================================================--

-- @mark SQL statement cache

--------------------------------------------------------------------------------

local function getStatement( self, sql )

	local cache = self._stmtCache

	local stmt = cache and cache[ sql ]

	if stmt then

		-- WARNING [scouten 2011-04-12]: If we ever start using APIs from
		-- WFSQLiteDatabaseImp that don't reset the statement upon completion,
		-- this needs to come back.

		--stmt:reset()

	else

		stmt = self._db:prepare( sql )
		cache[ sql ] = stmt

		-- If we haven't cached bindByPosition and gatherArrayOfArray yet,
		-- do it now. This saves us a bunch of overhead in callmethod of
		-- AgDirectObject.

		if not self._bindByPositionMethod then
			self._bindByPositionMethod = stmt:getMethod 'bindByPosition'
			self._runToCompletionMethod = stmt:getMethod 'runToCompletion'
			self._gatherArrayOfArrayMethod = stmt:getMethod 'gatherArrayOfArray'
			self._gatherArrayMethod = stmt:getMethod 'gatherArray'
		end

	end

	return stmt

end

--------------------------------------------------------------------------------

local function safeFinalize( stmt )

	local ok, maybeFinalizeError = pcall( stmt.finalize, stmt )

	if not ok then

		logger:errorf( "Failed to finalize statement: %s", maybeFinalizeError )

	end

end

--------------------------------------------------------------------------------

local function killStatement( self, sql )

	local cache = self._stmtCache

	if not cache then

		return

	end

	local stmt = cache[ sql ]

	if stmt then

		cache[ sql ] = nil

		safeFinalize( stmt )

	end

end

--------------------------------------------------------------------------------

local function isTableForIndexInSchema( db, indexName, schemaTables ) 
	
	local indexInfo = db:gatherArray(
		[[SELECT tbl_name FROM sqlite_master
			WHERE type == 'index'
				AND tbl_name != 'schemaVersion'
				AND sql NOT NULL
				AND name == ? ]], indexName )
	
	local tableName = indexInfo and indexInfo[ 1 ]
	
	-- return true if index's table is in the current set of schema tables
	return not not schemaTables[ tableName ]
	
end

--------------------------------------------------------------------------------

local function isColumnInTable( db, tableName, columnName )

	local tableColumnsSql = string.format( [[
		PRAGMA table_info( %s ) ]], tableName )

	local tableColumnsResult = db:gatherArrayOfArray( tableColumnsSql )
	killStatement( db, tableColumnsSql ) -- no need to cache statement

	if #tableColumnsResult > 0 then

		for _, column in ipairs( tableColumnsResult ) do
			if column[2] == columnName then
				return true
			end
		end

	end

	return false

end

--============================================================================--

-- @mark schema migration

--------------------------------------------------------------------------------

local function maybeInstallProgressHandlerForDocStoreInit( db, schema, existingIndexes )

		-- Inspect the missing indexes and the corresponding tables to estimate the
		-- effort of applying the missing indexes.
		-- The effort is measured in sqlite VM steps.

		local estimatedVMSteps = 0

		for indexName, indexDescription in pairs( schema._indexes ) do

			local existingIndexSql = existingIndexes[ indexName ]

			if not existingIndexSql or existingIndexSql ~= indexDescription._sql then
				local _, _, _, tableName = indexDescription._sql:find( "([%a_][%w_]*)%s+[Oo][Nn]%s+([%a_][%w_]*)" )

				-- Retrival of max(rowid) is fast compared to counting rows. Of course max(rowid)
				-- is not as good as estimator if the table contains deleted rows.
				-- But maxRow provides us with a suitable upper bound which is all we need.

				local maxRowSql = string.format( [[SELECT max(rowid) FROM "%s"]], tableName )
				local maxRowResult = db:gatherArray( maxRowSql )
				local maxRow = maxRowResult and maxRowResult[ 1 ]
				killStatement( db, maxRowSql )

				if maxRow then

					-- Indexes are B-trees, so construction will take O(n*log(n)),
					-- constant determined by experiment.

					estimatedVMSteps = estimatedVMSteps + 0.7 * maxRow * math.log(maxRow)
				end
			end

		end

		local count = 0
		local path = db:path()

		local function progressHandler()
			count = count + 1
			-- Cap progress count at 99% in case our effort estimation erred on the low side.
			WFCurrentThread.broadcastNotificationWithUserData( kDocStoreInitProgress,
				{ path, math.min( count, 99 ) } )
			return 0
		end

		-- Only install the progressHandler if the index creation will require substantial effort.
		if estimatedVMSteps > 5000 then
			db:setProgressCallback( progressHandler, math.floor( estimatedVMSteps/100 ) )
			-- send the initial progress status
			WFCurrentThread.broadcastNotificationWithUserData( kDocStoreInitProgress, { path, 0 } )
			return true
		end

		return false

end

--------------------------------------------------------------------------------

local function configureDatabaseForSchema( db, schema, path, progressHandlerRequested )

	-- Add required version info schema.

	schema:addTable [[CREATE TABLE schemaVersion( currentVersion, backwardCompatibleToVersion )]]

	-- IMPORTANT: the page_size pragma, if it exists, must appear prior to
	-- enabling WAL mode.

	if schema._pageSize then
		db:execOnce( "PRAGMA page_size = " .. tostring( schema._pageSize ) )
	end

	if schema._maxPageCount then
		db:execOnce( "PRAGMA max_page_count = " .. tostring( schema._maxPageCount ) )
	end

	if path:match( '^:memory:' ) then

		db:execOnce( "PRAGMA temp_store = memory" )

	else

		-- Turn on WAL support if so requested. (But don't when using in-memory
		-- databases, since that is kind of pointless and seems to hurt performance.)

		if schema._useWAL then
			db:execOnce( "PRAGMA journal_mode = WAL" )
		end

		if schema._synchronous then
			db:execOnce( "PRAGMA synchronous = " .. tostring( schema._synchronous ) )
		end

	end

	if schema._useForeignKeys then
		db:execOnce( "PRAGMA foreign_keys = true" )
	end

	-- Make sure all desired tables exist.

	local tableIntrospectionSql = [[SELECT tbl_name, sql FROM sqlite_master WHERE type == 'table']]

	local tableSchemaRows = db:gatherArrayOfArray( tableIntrospectionSql )
	killStatement( db, tableIntrospectionSql )

	local existingTables = {}

	for i = 1, #tableSchemaRows do

		local schemaRow = tableSchemaRows[ i ]
		existingTables[ schemaRow[ 1 ] ] = schemaRow[ 2 ]
			-- existingTables[ tblName ] = sql

	end

	local replaceAll = false

	for tableName, tableDescription in pairs( schema._tables ) do
					
		local existingTableSql = existingTables[ tableName ]

		if existingTableSql then

			-- Altering the table may generate a SQL string in which the whitespace
			-- does not match the whitespace in the schema definition. So, just to be 
			-- safe, we remove the whitespace from the strings before performing the
			-- comparison.

			existingTableSql = existingTableSql:gsub( "%s+", "" )
						
			local tableDescriptionSql = tableDescription._sql
			tableDescriptionSql = tableDescriptionSql and tableDescriptionSql:gsub( "%s+", "" )

			-- Table already exists in DB.

			if existingTableSql ~= tableDescriptionSql then

				-- Table exists and doesn't match desired schema.
				-- So far, we only know how to replace our way out of this problem.

				if tableDescription._onConflict == 'replaceAll' then
					replaceAll = true
					break
				end

				if tableDescription._onConflict == 'alter' and tableDescription._onConflictAlterSqlTable then

					-- Attempt to alter the conflicting table. Then recompare the
					-- SQL to see if the schemas match after the alter statement
					-- has been executed.

					for _, alterTableSql in ipairs( tableDescription._onConflictAlterSqlTable ) do

						-- if the ALTER statement contains an "add column" clause,
						-- check upfront if the column already exists in table;
						-- otherwise we get a sql exception which isn't fatal,
						-- but ugly and may mess up analytics
						local pattern = "[Aa][Dd][Dd]%s+[Cc][Oo][Ll][Uu][Mm][Nn]%s+(%a+)"
						local _, _, maybeAddColumn = string.find( alterTableSql, pattern )

						if not ( maybeAddColumn and isColumnInTable( db, tableName, maybeAddColumn ) ) then
							pcall( db.execOnce, db, alterTableSql )
							-- run the ALTER TABLE sql
						end

					end

					local singleTableIntrospectionSql = string.format( [[
							SELECT sql 
							FROM sqlite_master 
							WHERE type == 'table' and tbl_name == '%s' ]], tableName )

					local alteredTableResult = db:gatherArray( singleTableIntrospectionSql )
					killStatement( db, singleTableIntrospectionSql )
						
					local alteredTableSql = alteredTableResult[ 1 ]
					-- strip white space
					alteredTableSql = alteredTableSql:gsub( "%s+", "" )
					-- after renaming a table, the table name will be quoted; strip quotes
					alteredTableSql = alteredTableSql:gsub( '"', '' )
					
					if alteredTableSql == tableDescriptionSql then
					
						-- Success! The altered table SQL matches the schema we expected.
						break
						
					end

				end

				if tableDescription._onConflict ~= 'replace' then

					db._initErrorCode = initializationErrors.conflictingSchemaDefinitionError

					error( string.format(
						"WFSQLiteDatabase: conflicting definition for table %s\n\n%s",
						tableName, existingTableSql ), 2 )

				end

				-- Delete existing table and start over.

				db:execOnce( "DROP TABLE " .. tableName )
				db:execOnce( tableDescription._sql )

			end

		else

			-- No such table in DB: Create it now.

			db:execOnce( tableDescription._sql )

		end

	end

	if replaceAll then

		-- This is the nuclear option: Torch all content and start over.

		db:close()

		WFFileUtils.deleteFile( path )
		WFFileUtils.deleteFile( path .. '-shm' )
		WFFileUtils.deleteFile( path .. '-wal' )

		return 'restart'

	end

	-- Schema version check. (This might have to move ahead of the table
	-- creation phase above in future versions when we have a known incompatibility,
	-- but this will suffice to lock v1 clients out of future DBs and provide
	-- enough info for future clients to know they have to upgrade.)

	local currentVersion, backwardCompatibleToVersion =
						schema:currentAndBackwardCompatibleVersions()

	local schemaVersionCheckSql = [[SELECT currentVersion, backwardCompatibleToVersion
						FROM schemaVersion]]

	local schemaVersionRows = db:gatherArrayOfArray( schemaVersionCheckSql )
	killStatement( db, schemaVersionCheckSql )

	assert( schemaVersionRows and #schemaVersionRows <= 1,
			"WFSQLiteDatabase: bad response from schema query" )

	local schemaVersionRow = schemaVersionRows[ 1 ]
	if schemaVersionRow then

		-- WARNING [scouten 2015-02-04]: A bug that existed in this migration
		-- code from 2011 until present incorrectly interpreted the meaning
		-- of currentVersion. Test carefully with older versions of the software
		-- if doing any schema migration that involves the use of version flags.

		local db_backwardCompatibleVersion = schemaVersionRow[ 2 ]

		if type( db_backwardCompatibleVersion ) ~= 'string'
			or db_backwardCompatibleVersion > currentVersion
		then
			error( "WFSQLiteDatabase: database format is newer than current software" )
		end

		-- FIXME [scouten 2011-08-13]: In a future version, we'll need to
		-- upgrade the current/backward values to match the new values once
		-- we've upgraded the schema.

	else

		-- No pre-existing schema statement: Probably a new DB..

		db:execOnce( [[INSERT INTO schemaVersion VALUES( ?, ? )]], currentVersion,
							backwardCompatibleToVersion )

	end

	for _, tableToDrop in ipairs( schema._tablesToDrop ) do
		db:execOnce( "DROP TABLE IF EXISTS " .. tableToDrop )
	end
	
	-- Make sure all desired indexes exist.

	local indexIntrospectionSql = [[SELECT name, sql FROM sqlite_master WHERE type == 'index']]

	local indexSchemaRows = db:gatherArrayOfArray( indexIntrospectionSql )
	killStatement( db, indexIntrospectionSql )

	local existingIndexes = {}

	for i = 1, #indexSchemaRows do

		local schemaRow = indexSchemaRows[ i ]
		existingIndexes[ schemaRow[ 1 ] ] = schemaRow[ 2 ]
			-- existingIndexes[ tblName ] = sql

	end

	-- Possibly install a progress handler for tracking the creation of missing indexes.

	local isProgressHandlerInstalled = false

	if progressHandlerRequested then
		isProgressHandlerInstalled = maybeInstallProgressHandlerForDocStoreInit( db, schema, existingIndexes )
	end

	for indexName, indexDescription in pairs( schema._indexes ) do

		local existingIndexSql = existingIndexes[ indexName ]

		if existingIndexSql then

			-- Index already exists in DB.

			if existingIndexSql ~= indexDescription._sql then

				-- Index exists and doesn't match desired schema.
				-- Unlike tables, we don't bother with a conflict strategy;
				-- we always blow away any unwanted index and start over.

				db:execOnce( "DROP INDEX " .. indexName )
				db:execOnce( indexDescription._sql )

			end

		else

			-- No such index in DB: Create it now.

			db:execOnce( indexDescription._sql )

		end

	end

	if isProgressHandlerInstalled then

		db:setProgressCallback( nil )

		-- send final progress notification
		WFCurrentThread.broadcastNotificationWithUserData( kDocStoreInitProgress, { path, 100 } )
	end

	-- If a sqlite progress handler was requested always send a final 'done' message so
	-- that clients can properly close progress streams etc.
	if progressHandlerRequested then

		WFCurrentThread.broadcastNotificationWithUserData( kDocStoreInitProgress, { path, 'done' } )

	end

	-- Remove unwanted indexes only for tables that are in the current schema. If the table isn't
	-- in the current set of schema tables, we don't own it so we shouldn't delete it. 
	for indexName, _ in pairs( existingIndexes ) do
		if not schema._indexes[ indexName ]  
			and isTableForIndexInSchema( db, indexName, schema._tables ) then
			
			db:execOnce( "DROP INDEX " .. indexName )
		
		end
	end

end

--============================================================================--

-- @mark WFSQLiteDatabase class

--------------------------------------------------------------------------------
--[[!
	\class WFSQLiteDatabase

	\brief This namespace provides direct access to an SQLite database.

	\sa WFSQLiteDatabaseSchema

	A worker thread's init script defines a schema by creating a \c WFSQLiteDatabaseSchema
	object, adding the appropriate table and index definitions, and then converts
	it to a running database by calling:

	<pre>
		 WFSQLiteDatabase( schema, ... )
	</pre>

	Use the additional arguments to pass through the desired database path.

	This call opens the SQLite database connection, ensures that the database schema
	matches what was specified in the schema declaration.

	The call returns a reference to the database object, which represents this
	database connection; the thread can use it for direct access to the database
	if needed.

	Once this call completes, the database schema is frozen for the lifetime of the
	thread. Any change you make to the database schema object are ignored for this
	database instance.
]]--

--------------------------------------------------------------------------------

local databaseMethods = {

	_db = false,
		-- (WFSQLiteDatabaseImp)

	_stmtCache = false,
		-- (table) cache of prepared statements
			-- key: (string) SQL to compile
			-- value: (WFSQLiteStatement)

	_path = false,
		-- (string) path to the database

	_hasActiveTransaction = false,
		-- (boolean) true if we're in the midst of a :withTransactionDo block

	_bindByPositionMethod = false,
		-- (function) optimized bypass for bindByPosition method of WFSQLiteStatement

	_runToCompletionMethod = false,
		-- (function) optimized bypass for runToCompletion method of WFSQLiteStatement

	_gatherArrayOfArrayMethod = false,
		-- (function) optimized bypass for gatherArrayOfArray method of WFSQLiteStatement

	_gatherArrayMethod = false,
		-- (function) optimized bypass for gatherArray method of WFSQLiteStatement

	_nTransactions = 0,
		-- (number) how many transactions have been issued on this DB?

	_autosaveInterval = false,
		-- (number) if present, this is the maximum number of seconds to allow between
		-- actual transactions; savepoints are used for interim changes within that
		-- interval

	_autosaveTimerDisposable = false,
		-- (Rx.Disposable) if present, this handle can be used to cancel the timer
		-- action that has been scheduled

	_initErrorCode = false,
		-- (number) an error code that is assigned when db initialization fails.

	_sqliteFatalErrorCodeStream = false,
		-- (Observable) an observable that fires with SQLite error codes when a database
		-- operation encounters a critical error.

	_PERF_gatherStats = false,
		-- (boolean) true to gather performance data about queries run against
		-- this database

	_PERF_stmtStats = false,
		-- (table) stats for each statement
			-- key: (string) SQL from statement
			-- value: (table) info on statement (count, min, max, total)

	_PERF_queriesInTransaction = false,
		-- (table) while in a transaction, this accumulates the queries that were executed
			-- in this transaction

	_log = false,
		-- (WFLogger)

	_logInfo = false,
		-- (boolean) true if log level is DEBUG or INFO

	_logDebug = false,
		-- (boolean) true if log level is DEBUG

	WFCloseSupport.addCloseSupportMethods,

}

--------------------------------------------------------------------------------

local dbInstanceCounter = 1

--------------------------------------------------------------------------------

local function logInfo( self, ... )

	if self._logInfo then
		self._log:infof( ... )
	end

end

--------------------------------------------------------------------------------

local function logDebug( self, ... )

	if self._logDebug then
		self._log:debugf( ... )
	end

end

--------------------------------------------------------------------------------

local function makeReadOnlyNativeDatabase( maybeSpyDb, path, useWAL )

	if maybeSpyDb then

		return maybeSpyDb

	end

	if not path then

		error( "must provide path" )

	end

	local db = WFSQLiteDatabaseImp.makeReadOnlyDatabase( path )

	if not db then

		error( string.format( "WFSQLiteDatabase.makeDatabase( %s ) failed",
								tostring( path ) ) )

	end
	
	return db

end

--------------------------------------------------------------------------------

local function makeNativeDatabase( maybeSpyDb, path, schema )

	if maybeSpyDb then

		return maybeSpyDb

	end

	if not path then

		error( "must provide path" )

	end

	local db = WFSQLiteDatabaseImp.makeDatabase( path )

	if not db then

		error( string.format( "WFSQLiteDatabase.makeDatabase( %s ) failed",
								tostring( path ) ) )

	end

	db = WFSQLiteCorruptionLogger.addDb( db )
	
	if schema._allowMultipleConnections then
		db:exec( "PRAGMA locking_mode=NORMAL" )
	else
		if schema._useWAL and not disableExclusiveLocking then
			-- 2017.11.20 bkruse 
			-- 
			-- As of SQLite 3.19.1, there is strong circumstantial evidence that
			-- combination of WAL and the memory mapped SHM file increases the
			-- likelihood of corruption if the app aborts in the middle of a SQLite 
			-- write operation. This is based both on LrCC corruption reports in the
			-- wild as well as reports to the SQLite mailing list.
			-- 
			-- The suggested workaround is to disable memory mapped files. There 
			-- are a number of ways of doing this, including compile time switches.
			-- 
			-- Another method is this PRAGMA below which must be entered before any
			-- other access happens on the catalog
			db:exec( "PRAGMA locking_mode=EXCLUSIVE" )
		end
	end

	return db

end

--------------------------------------------------------------------------------
--[[
	\brief Creates a \c WFSQLiteDatabase instance.

	\param path (string) path where the database should be created
		Can be special strings \c ":memory:" or \c ":temp:" which will create
		an in-memory or on-disk temporary database.
	\param schema (\c WFSQLiteDatabaseSchema) schema for the database
	\param options (optional, table) configuration options

		<ul>
			<li> \c readOnly (boolean): True to open the database as read only
			<li> \c logLevel (string): WFLogger log level to set for the 
				WFSQLiteDatabase logging
			<li> \c disableQueryLogging (boolean): True to improve performance
				by not logging query performance for diagnostic log.</li>
		</ul>
]]--

function databaseMethods:__init( path, schema, options )

	check( 'WFSQLiteDatabase', 'string|databaseImpSpy|userdata',
		   'WFSQLiteDatabaseSchema', '?table|boolean' )

	if not options or type( options ) == 'boolean' then
		options = {
			disableQueryLogging = not not options
		}
	end

	self._log = not useSharedLogger and
					WFLogger( string.format( "WFSQLiteDatabase %s %d in thread %s", path,
										 dbInstanceCounter,
										 WFCurrentThread.DEBUG_getThreadName() ) ) or
					logger


	if options.logLevel then
		self._log:setLevel( options.logLevel )
	end

	dbInstanceCounter = dbInstanceCounter + 1

	-- For testing purposes only, we allow calling code to pass in an
	-- object to use in place of the native database imp.

	local spyDb

	if type( path ) == 'table' then
		spyDb = path
		path = spyDb._path
	end

	-- Translate local special variants of path into SQLite versions.

	if type( path ) == 'string' then

		if path:sub( 1, 8 ) == ':memory:' then

			path = ':memory:'

		elseif path:sub( 1, 6 ) == ':temp:' then

			-- Per http://www.sqlite.org/inmemorydb.html, this will cause an
			-- on-disk DB to be created and then destroyed when the connection
			-- is destroyed.

			path = ''

		end

	end

	self._logInfo = self._log:will_info()
	self._logDebug = self._log:will_debug()


	if self._logInfo then
		self._log:info( "Make database" )
	end
	
	if options.readOnly then
		self._db = makeReadOnlyNativeDatabase( spyDb, path )
	else
		self._db = makeNativeDatabase( spyDb, path, schema )
	end
	
	path = type( path ) == "string" and path or self._db:path()

	self._path = path

	self._stmtCache = {}

	-- Collect diagnostic log data?

	if COLLECT_DIAGNOSTIC_LOG_DATA and not options.disableQueryLogging then

		self:PERF_collectQueryPerformanceStats()

		WFDiagnosticLog.logDataRequestE()
			:ownedOnDataE( self, function()
				self:sendDiagnosticLogData()
			end )

	end
	
	local installProgressHandler = options and options.installProgressHandlerForDocStoreInit or false
	
	if not options.readOnly then

		-- Ensure that we have desired schema in place.
		
		local success, message = pcall( function()
			while true do
	
				local result = configureDatabaseForSchema( self, schema, path, installProgressHandler )
	
				if result == 'restart' then
					self._db = makeNativeDatabase( spyDb, path, schema )
					self._stmtCache = {}
				else
					break
				end
	
			end
		end )

		if not success then
	
			if not self._initErrorCode then
				self._initErrorCode = initializationErrors.standardError
			end
			
			if string.find( message, "error code 14" ) then
				local KSPathUtils = require 'KSPathUtils'
	
				-- This logic mirrors how SQLite finds where to store temp files.
				local tempDirs = {
					os.getenv( "SQLITE_TMPDIR" ) or false,
					os.getenv( "TMPDIR" ) or false,
					"/var/tmp",
					"/tmp"
				}
				local addlMessages = { message, }
				for _, dir in ipairs( tempDirs ) do
					if dir and WFFileUtils.exists( dir ) then
						local fileSystemFreeSpace, fileSystemSize
								= WFFileUtils.freeSpaceAtPath( dir )
						local testFile = KSPathUtils.child( dir, "hack_test_filesystem.txt" )
						testFile = WFFileUtils.ensureFileNameIsUnique( testFile )
						local f, errMsg = io.open( testFile, "w" )
						if f then
							local writeResult = f:write( string.rep( "four", 1024 ) )
							errMsg = "Able to open temp file for writing and write returned .. " .. tostring( writeResult )
							f:close()
							WFFileUtils.delete( testFile )
						end
	
						addlMessages[ #addlMessages + 1 ] = string.format( "%s dir exists with %u bytes free of %u total (%s)", dir, fileSystemFreeSpace, fileSystemSize, errMsg or "No info from file test" )
					end
				end
				message = table.concat( addlMessages, "\n" )
			end
	
			-- If we fail here, DB is probably corrupt. Close it out now.
	
			self._db:close()
			self._db = nil
	
			message = string.format( "%s :: Error Code: %s", message, self._initErrorCode or "Unknown" )
	
			error( message )
	
		end
	end

	self._sqliteFatalErrorCodeStream = Rx.ReplaySubject( 1 )
	self._sqliteFatalErrorCodeStream:onNext( false )

end

--------------------------------------------------------------------------------

local function assertDatabaseNotClosed( self )

	local db = self._db

	if not db then
		self._log:errorf( "Database used after it was closed via %s", ':traceback:' )
		error( "WFSQLiteDatabase: database was closed", 3 )
	end

	return db

end

--------------------------------------------------------------------------------
--[[!
	\brief Closes this database connection.
]]--

local function closeImp( self, closeAPI )

	check( 'WFSQLiteDatabase' )

	if self._autosaveTimerDisposable then

		local ok, maybeSaveErr = pcall( self.save, self )

		if not ok then

			logger:errorf( "failure to save on close: %s", maybeSaveErr )

		end

	end

	if self._db then

		local db = self._db

		self._db = false

		logInfo( self, "Finalizing statements in preparation for closing database" )

		local stmtCache = self._stmtCache

		self._stmtCache = false

		if stmtCache then

			for _, stmt in pairs( stmtCache ) do

				safeFinalize( stmt )

			end

		end

		logInfo( self, "Closing database" )

		local ret = db[ closeAPI ]( db )

		return ret

	end

end

--------------------------------------------------------------------------------

function databaseMethods:close()

	return closeImp( self, "close" )

end

--------------------------------------------------------------------------------

function databaseMethods:createTransitableRefAndClose()

	return closeImp( self, "createTransitableRefAndClose" )

end

--------------------------------------------------------------------------------
--[[!
	\brief Retrieves the disk path to this database file.

	\return (string) The path.
]]--

function databaseMethods:path()

	check( 'WFSQLiteDatabase' )

	return self._path

end

--------------------------------------------------------------------------------

--------------------------------------------------------------------------------

--[[!
	\brief Simuluates corruption on underlying Db

]]--

--------------------------------------------------------------------------------

local function makePassThruMethod( realObject, methodName )

	return function( _, ... )

		return realObject[ methodName ]( realObject, ... )

	end

end

--------------------------------------------------------------------------------

local function makeSimulatedCorruptNativeSqliteStmt( realNativeSQLiteStmt )

	local simulatedCorruptNativeSQLiteStmt = {

		_realNativeSQLiteStmt = realNativeSQLiteStmt

	}

	local function alwaysRaisesCorruptError()

		realNativeSQLiteStmt:_DEBUG_SIMULATE_CORRUPT_ERROR()

	end

	local _passThruMethods = {}
	
	setmetatable( 
		simulatedCorruptNativeSQLiteStmt, 
		{

			__index = 
				function( _, methodName )

					-- override any methods that will read sqlite db pages 
					-- and have them act as if corruption occurred
					if methodName == "step" or
							methodName == "stepValues" or
							methodName == "runToCompletion" or
							methodName == "gatherArray" or
							methodName == "gatherArrayOfTables" or
							methodName == "gatherMap" or
							methodName == "gatherArrayOfArray" or
							methodName == "gatherSet" then

						return alwaysRaisesCorruptError

					end

					-- for any other native db methods, just act as normal

					_passThruMethods[ methodName ] =
						_passThruMethods[ methodName ] or 
						makePassThruMethod( realNativeSQLiteStmt, methodName )

					return _passThruMethods[ methodName ]

				end,

		} )

	return simulatedCorruptNativeSQLiteStmt

end

--------------------------------------------------------------------------------
-- makeSimulatedCorruptNativeDb is supposed to have same interface as
-- db returned makeNativeDatabase( ... ) except that it will raise corruption
-- errors if calling a db method that would read sqlite pages
--------------------------------------------------------------------------------

local function makeSimulatedCorruptNativeDb( realNativeDb )

	local simulatedCorruptNativeDb = {}

	local function alwaysRaisesCorruptError()

		realNativeDb:exec( "SELECT test_corrupt_db_error()" )

	end

	local _passThruMethods = {}

	local function prepareCorruptSQLiteStmt( _, ... )

		local stmt = realNativeDb:prepare( ... )

		return makeSimulatedCorruptNativeSqliteStmt( stmt )

	end
	
	setmetatable( 
		simulatedCorruptNativeDb, 
		{

			__index = 
				function( _, methodName )

					-- override any methods that will read sqlite db pages 
					-- and have them act as if corruption occurred
					if methodName == "onlineBackup" or 
							methodName == "exec" or
							methodName == "execOnce" then

						return alwaysRaisesCorruptError


					end

					-- override methods that make sqlite stmts into one that
					-- returns sqlite statement that emit corrupt errors

					if methodName == "prepare" then

						return prepareCorruptSQLiteStmt

					end

					-- for any other native db methods, just act as normal

					_passThruMethods[ methodName ] =
						_passThruMethods[ methodName ] or 
						makePassThruMethod( realNativeDb, methodName )

					return _passThruMethods[ methodName ]

				end,

		} )

	return simulatedCorruptNativeDb

end

--------------------------------------------------------------------------------

local function bindByPositionForSimulatedCorruptSQLiteStmt( simulatedCorruptSQLiteStmt, ... )

	local stmt = simulatedCorruptSQLiteStmt._realNativeSQLiteStmt

	return stmt:bindByPosition( ... )

end

--------------------------------------------------------------------------------

local function raiseCorruptionErrorOnNativeSQLiteStmt( simulatedCorruptSQLiteStmt )

	local stmt = simulatedCorruptSQLiteStmt._realNativeSQLiteStmt

	return stmt:_DEBUG_SIMULATE_CORRUPT_ERROR()

end

--------------------------------------------------------------------------------

local function replaceStmtMethodSelfCacheWithCorruptors( self )

	self._bindByPositionMethod = bindByPositionForSimulatedCorruptSQLiteStmt

	self._runToCompletionMethod = raiseCorruptionErrorOnNativeSQLiteStmt

	self._gatherArrayOfArrayMethod = raiseCorruptionErrorOnNativeSQLiteStmt

	self._gatherArrayMethod = raiseCorruptionErrorOnNativeSQLiteStmt

end

--------------------------------------------------------------------------------

function databaseMethods:_DEBUG_simulateCorruption()

	check( 'WFSQLiteDatabase' )

	local nativeDb = self._db

	self._db = makeSimulatedCorruptNativeDb( nativeDb )

	local newStmtCache = {}

	local didReplaceStmtMethodSelfCache

	for sql, stmt in pairs( self._stmtCache ) do

		local simulatedCorruptSQLiteStmt = makeSimulatedCorruptNativeSqliteStmt( stmt )

		if not didReplaceStmtMethodSelfCache then

			didReplaceStmtMethodSelfCache = true

			replaceStmtMethodSelfCacheWithCorruptors( self )

		end

		newStmtCache[ sql ] = simulatedCorruptSQLiteStmt

	end
	
	if not didReplaceStmtMethodSelfCache and self._bindByPositionMethod then

		replaceStmtMethodSelfCacheWithCorruptors( self )
		
	end

	self._stmtCache = newStmtCache

end

--------------------------------------------------------------------------------

--[[!
	\brief Simuluates corruption on underlying Db

]]--

function databaseMethods:_DEBUG_simulateIOError()

	check( 'WFSQLiteDatabase' )

	self._db:exec( "SELECT test_transient_io_error()" )

end

--------------------------------------------------------------------------------

local sqlContinuationIndent = "\n" .. string.rep( " ", 48 )

--------------------------------------------------------------------------------

local queryStartTime

local function finish_profileQuery( self, sql, ... )

	local elapsedTime = debug.timeNow() - queryStartTime

	if ( sql == 'COMMIT' or sql:match( '^RELEASE ' ) )
		and self._PERF_queriesInTransaction
	then

		local t = {}

		for k in pairs( self._PERF_queriesInTransaction ) do
			t[ #t + 1 ] = k
		end

		table.sort( t )

		local t2 = {}

		for i, k in pairs( t ) do
			local n = self._PERF_queriesInTransaction[ k ]
			if n > 1 then
				t2[ i ] = string.format( "(%d times) : %s", n, k )
			else
				t2[ i ] = k
			end
		end

		sql = 'COMMIT :: queries were:' .. sqlContinuationIndent .. table.concat( t2, sqlContinuationIndent )
		self._PERF_queriesInTransaction = false

	elseif self._hasActiveTransaction and self._PERF_queriesInTransaction and sql ~= 'BEGIN TRANSACTION' then

		if not self._PERF_queriesInTransaction[ sql ] then
			self._PERF_queriesInTransaction[ sql ] = 1
		else
			self._PERF_queriesInTransaction[ sql ] = self._PERF_queriesInTransaction[ sql ] + 1
		end

	end

	local stat = self._PERF_stmtStats[ sql ]

	if stat then
		stat.count = stat.count + 1
		stat.min = math.min( stat.min, elapsedTime )
		stat.max = math.max( stat.max, elapsedTime )
		stat.total = stat.total + elapsedTime
	else
		stat = {
			sql = sql,
			count = 1,
			min = elapsedTime,
			max = elapsedTime,
			total = elapsedTime,
		}
		self._PERF_stmtStats[ sql ] = stat
	end

	self._PERF_gatherStats = true

	return ...

end

--------------------------------------------------------------------------------

local function profileQuery( self, sql, methodName, ... )

	self._PERF_gatherStats = false

	queryStartTime = debug.timeNow()

	return finish_profileQuery( self, sql, self[ methodName ]( self, sql, ... ) )

end

--------------------------------------------------------------------------------

local function processSQLiteErrorAndRethrow( self, ... )

	local errMsg = ...
	local errCode

	-- Extract the error code from the error message. Error message of the form:
	-- Database \"%s\": %s\n  sql: %s (error code %d)
	-- See WFSQLite_checkError within WFSQLiteExceptions.c for more info.

	if type( errMsg ) == 'string' then

		local match = errMsg:match( "error code (%d+)" )

		if match then
			errCode = tonumber( match )
		end

	end

	-- Check if the error code represents a critical issue, potentially
	-- requiring user intervention.

	if errCode and (
			errCode == SQLITE_IOERR_ERROR or
			errCode == SQLITE_FULL_ERROR or
			errCode == SQLITE_CORRUPT_ERROR ) then

		self._sqliteFatalErrorCodeStream:onNext( errCode )

	end

	error( ... )

end

--------------------------------------------------------------------------------

local function finish_execOnce( self, sql, success, ... )

	if success then

		killStatement( self, sql )
		return ...

	else

		processSQLiteErrorAndRethrow( self, ... )

	end

end

--------------------------------------------------------------------------------
--[[!
	\brief Executes an SQL query exactly once.

	\param sql (string) The query.
	\param ... Any additional parameters needed for the query.

	\return Multiple results, with each result representing one row as an array.
		Because this is limited by the size of the Lua stack, you should
		use \c :gatherArrayOfArray for a query that can return more than a few rows.
]]--

function databaseMethods:execOnce( sql, ... )

	check( 'WFSQLiteDatabase', '+string' )

	logInfo( self, "execOnce %s", sql )

	local stmt = getStatement( self, sql )
	stmt:bindByPosition( ... )

	return finish_execOnce( self, sql, pcall( stmt.runToCompletion, stmt ) )

end

--------------------------------------------------------------------------------
--[[!
	\brief Prepare and bind a statement

	This can be used for stepping through a result set.
	
	Don't forget to finalize the statement!

	\param sql (string) The query.
	\param ... Any additional parameters needed for the query.

	\return statement
]]--

function databaseMethods:prepare( sql, ... )

	local stmt = self._db:prepare( sql )
	stmt:bindByPosition( ... )

	return stmt
end

--------------------------------------------------------------------------------

local function finish_exec( self, success, ... )

	if success then
		return ...
	else
		processSQLiteErrorAndRethrow( self, ... )
	end

end

--------------------------------------------------------------------------------
--[[!
	\brief Executes an SQL query and caches the query so that it executes
		faster on subsequent calls.

	\param sql (string) The query.
	\param ... Any additional parameters needed for the query.

	\return Multiple results, with each result representing one row as an array.
		Because this is limited by the size of the Lua stack, you should
		use \c :gatherArrayOfArray for a query that can return more than a few rows.
]]--

function databaseMethods:exec( sql, ... )

	check( 'WFSQLiteDatabase', '+string' )

	if self._PERF_gatherStats then
		return profileQuery( self, sql, 'exec', ... )
	end

	logInfo( self, "exec %s", sql )

	local stmt = getStatement( self, sql )
	self._bindByPositionMethod( stmt, ... )

	return finish_exec( self, pcall( self._runToCompletionMethod, stmt ) )

end

--------------------------------------------------------------------------------
--[[!
	\brief Executes an SQL query and returns the result as an array of arrays.

	If your query generates a single column per row, consider using
	\c :gatherArray instead as it will be more efficient.

	\param sql (string) The query.
	\param ... Any additional parameters needed for the query.

	\return The result as an array of arrays, where each contained array
		represents one row.
]]--

function databaseMethods:gatherArrayOfArray( sql, ... )

	check( 'WFSQLiteDatabase', '+string' )

	if self._PERF_gatherStats then
		return profileQuery( self, sql, 'gatherArrayOfArray', ... )
	end

	logInfo( self, "gatherArrayOfArray %s", sql )

	local stmt = getStatement( self, sql )
	self._bindByPositionMethod( stmt, ... )

	return self._gatherArrayOfArrayMethod( stmt )

end

--------------------------------------------------------------------------------
--[[!
	\brief Executes an SQL query and returns a mixed table with the checksum and the rows as arrays.

		
	A result with two rows of data would look like this:
	<pre>
	{ 
		sha256 = '0db078',
		{
			{ 'a, b' },
			{ 'c', 'd' }
			
		}
	}
	</pre>

	\param sql (string) The query.
	\param ... Any additional parameters needed for the query.

	\return The result as a a mixed table with the checksum and the rows as arrays.
]]--

function databaseMethods:gatherArrayOfArrayWithChecksum( sql, ... )

	check( 'WFSQLiteDatabase', '+string' )

	if self._PERF_gatherStats then
		return profileQuery( self, sql, 'gatherArrayOfArrayWithChecksum', ... )
	end

	logInfo( self, "gatherArrayOfArrayWithChecksum %s", sql )

	local stmt = getStatement( self, sql )
	self._bindByPositionMethod( stmt, ... )

	return stmt:gatherArrayOfArrayWithChecksum()

end


--------------------------------------------------------------------------------
--[[!
	\brief Executes an SQL query and returns the result as an array.

	IMPORTANT: This method is optimized specifically for queries that return a
	single column for each row. Any values after the first column will be
	discarded.

	\param sql (string) The query.
	\param ... Any additional parameters needed for the query.

	\return The result as an array where each entry is the column value
]]--

function databaseMethods:gatherArray( sql, ... )

	check( 'WFSQLiteDatabase', '+string' )

	if self._PERF_gatherStats then
		return profileQuery( self, sql, 'gatherArray', ... )
	end

	logInfo( self, "gatherArray %s", sql )

	local stmt = getStatement( self, sql )
	self._bindByPositionMethod( stmt, ... )

	return self._gatherArrayMethod( stmt )

end

--[[!
	\brief Executes an SQL query and returns a mixed table with the checksum and the result values.

	IMPORTANT: This method is optimized specifically for queries that return a
	single column for each row. Any values after the first column will be
	discarded.
	
	A result with two rows of data would look like this:
	<pre>
	{ 
		sha256 = '0db078',
		'a',
		'b'
	}
	</pre>

	\param sql (string) The query.
	\param ... Any additional parameters needed for the query.

	\return The result as a a mixed table with the checksum and the rows as arrays.
]]--

function databaseMethods:gatherArrayWithChecksum( sql, ... )

	check( 'WFSQLiteDatabase', '+string' )

	if self._PERF_gatherStats then
		return profileQuery( self, sql, 'gatherArrayWithChecksum', ... )
	end

	logInfo( self, "gatherArrayWithChecksum %s", sql )

	local stmt = getStatement( self, sql )
	self._bindByPositionMethod( stmt, ... )

	return stmt:gatherArrayWithChecksum()

end

--------------------------------------------------------------------------------

local function maybeCallReducer( accumVal, reducer, ... )

    if select( '#', ... ) == 0 then

        return false, accumVal

    end

    return true, reducer( accumVal, ... )

end

--------------------------------------------------------------------------------

local function doSQLQueryReductionWithValues( stmt, initAccumVal, reducer )

    local shouldStep, accumVal = 
            maybeCallReducer( initAccumVal, reducer, stmt:stepValues() )

    while shouldStep do

        shouldStep, accumVal = 
            maybeCallReducer( accumVal, reducer, stmt:stepValues() )

    end

    return accumVal

end

--------------------------------------------------------------------------------

local function maybeReraise( ok, maybeErr )
	if not ok then
		logger:error( maybeErr )
		error( maybeErr )
	end
end

--------------------------------------------------------------------------------

local function makeReduceMethod( runReducerFunc )
	return function( self, initAccumVal, reducer, sql, ... )
		assert( type( reducer ) == "function", "Second argument to reduceSQLQuery must be a reducer function." )
		assert( type( sql ) == "string", "Third argument to reduceSQLQuery must be a SQL statement." )

		local stmt = getStatement( self, sql )
		if select( "#", ... ) > 0 then
			local ok, maybeMsg = pcall( stmt.bindByPosition, stmt, ... )
			if not ok then
				maybeReraise( ok, maybeMsg )
			end
		end
		local ok, accumVal = pcall( runReducerFunc, stmt, initAccumVal, reducer )
		maybeReraise( ok, accumVal )
		
		stmt:reset()
		return accumVal
	end
end

--------------------------------------------------------------------------------
-- reduceSQLQueryViaValueArrays is not going to work if the value arrays have nulls
--------------------------------------------------------------------------------

databaseMethods.reduceSQLQueryViaValueArrays = makeReduceMethod( doSQLQueryReductionWithValues )


--------------------------------------------------------------------------------
--[[!
	\brief Executes an SQL query and returns the result as an encoded by a custom
		native encoder.

	\param sql (string) The query.
	\param encoder (function) The custom encoder
	\param ... Any additional parameters needed for the query.

	\return The result in whatever form the encoder created it.
]]--

function databaseMethods:encodeResults( sql, encoder, ... )

	check( 'WFSQLiteDatabase', '+string', 'function' )

	if self._PERF_gatherStats then
		return profileQuery( self, sql, 'encodeResults', encoder, ... )
	end

	self._log:infof( "encodeResults %s", sql )

	local stmt = getStatement( self, sql )
	self._bindByPositionMethod( stmt, ... )

	local result = encoder( stmt )
	stmt:reset()
	return result

end

--------------------------------------------------------------------------------
--[[!
	\brief Retrieves the most recently inserted row.

	\return The row ID.
]]--

function databaseMethods:getLastInsertRowId()

	check( 'WFSQLiteDatabase' )

	logInfo( self, "getLastInsertRowId" )

	local db = assertDatabaseNotClosed( self )
	return db:getLastInsertRowID()

end

--------------------------------------------------------------------------------
--[[!
	\brief Return the number of affected rows.

]]--

function databaseMethods:getAffectedRowCount()

	check( 'WFSQLiteDatabase' )

	logInfo( self, "getAffectedRowCount" )

	local db = assertDatabaseNotClosed( self )
	return db:getAffectedRowCount()
	
end

--------------------------------------------------------------------------------
--[[!
	\brief Sets the change counter for this connection.

	\param changeCounter (number) The new value.
]]--

function databaseMethods:setChangeCounter( changeCounter )

	check( 'WFSQLiteDatabase', 'number' )

	logInfo( self, "setChangeCounter %s", changeCounter )

	local db = assertDatabaseNotClosed( self )
	db:setChangeCounter( changeCounter )

end

--------------------------------------------------------------------------------
--[[!
	\brief Retrieves the change counter for this connection.

	\return (number) The current counter value.
]]--

function databaseMethods:getChangeCounter()

	check( 'WFSQLiteDatabase' )

	logInfo( self, "getChangeCounter" )

	local db = assertDatabaseNotClosed( self )
	return db:getChangeCounter()

end


--[[!
	\brief Set a progress callback.
	
	Install a callback to be repeatedly called during a long-running sqlite operation.
	
	If the callback returns a nonzero value, the ongoing sqlite operation will be cancelled.
	
	\param cb The progress callback.
	\param instructions The approximate number of sqlite virtual machine instructions 
		that are evaluated between successive invocations of the callback.

]]--

function databaseMethods:setProgressCallback( callback, instructions )

	check( 'WFSQLiteDatabase' )

	logInfo( self, "setProgressCallback" )

	local db = assertDatabaseNotClosed( self )
	return db:setProgressCallback( callback, instructions )

end

--------------------------------------------------------------------------------
--[[
	Intended semantics for savepoint logic state variables:

	"outer savepoint": Savepoint transaction that actually gets committed to disk.

	"inner savepoint": Savepoint transactions created inside the outer savepoint
		that can be used to rollback the results of individual logic functions in
		the event of failure.

	"logic function": The function provided as an argument to :withTransactionDo.

	_hasActiveTransaction: true if we're currently running the logic function

	_autosaveTimerDisposable: present if we have successfully completed at least
		one logic function (this is an unsubscribe handle to trigger the save to
		commit the outer savepoint)
]]--

--------------------------------------------------------------------------------

local function finish_withSavepointDo( self, success, ... )

	logInfo( self, "finish_withSavepointDo success %s", success )

	self._hasActiveTransaction = false

	if success then

		-- Commit inner savepoint.

		logDebug( self, "Release inner savepoint" )
		self:exec 'RELEASE inner'

		-- If we don't have any open savepoints, then start timer to auto-save.

		if not self._autosaveTimerDisposable then

			self._autosaveTimerDisposable = Observable.timer( self._autosaveInterval, WFRxScheduler.sharedScheduler() )
				:subscribe( function()
					self:save()
					self._autosaveTimerDisposable = nil
				end )

		end

		-- Pass along logic func results to caller.

		return ...

	else

		-- Roll back to inner savepoint.

		logDebug( self, "Logic func failed: rolling back inner savepoint" )
		self:exec 'ROLLBACK TO inner'

		-- If that's the only open savepoint, also roll back to outer savepoint.

		if not self._autosaveTimerDisposable then
			logDebug( self, "Also rolling back outer savepoint" )
			self:exec 'ROLLBACK TO outer'
		end

		-- Replay error for caller.

		error( ... )

	end

end

--------------------------------------------------------------------------------

local function withSavepointDo( self, func, ... )

	if not self._autosaveTimerDisposable then
		logDebug( self, "Starting new outer savepoint" )
		self:exec 'SAVEPOINT outer'
	end

	logDebug( self, "Starting new inner savepoint" )
	self:exec 'SAVEPOINT inner'

	return finish_withSavepointDo( self, pcall( func, ... ) )

end

--------------------------------------------------------------------------------

local function noopFn()
	return
end

--------------------------------------------------------------------------------

local function finish_withTransactionDo( self, maybeOnTransactionFinished, success, ... )

	logInfo( self, "finish_withTransactionDo success %s", success )

	self._hasActiveTransaction = false

	maybeOnTransactionFinished = maybeOnTransactionFinished or noopFn

	if success then
	
		pcall( self.exec, self, "COMMIT" )
		maybeOnTransactionFinished()
		return ...
		
	else
		
		pcall( self.exec, self, "ROLLBACK" )
		maybeOnTransactionFinished()
		error( ... )
		
	end

end

--------------------------------------------------------------------------------
--[[!
	\brief Returns true if db is running in a transaction, use this to
		determine if it's safe to call \c withTransactionDo

	\return boolean if transaction is running
]]--

function databaseMethods:hasActiveTransaction()

	return self._hasActiveTransaction

end

--------------------------------------------------------------------------------
--[[!
	\brief Applies a function within a single database transaction.

	<ul><li>Do not call this from within a function that is already being called by
		\c withTransactionDo (transactions cannot be nested). </li>

	<li>You cannot use \c coroutine.yield within this call.</li></ul>

	\param func The function, which calls functions such as \c :exec to execute
		SQL queries within the transaction, and returns normally to commit.
		If the function returns an error, the transaction is rolled back.

	\param ... Any additional parameters for the function.

	\return The result.
]]--

function databaseMethods:withTransactionDo( func, ... )

	check( 'WFSQLiteDatabase', 'function' )

	logInfo( self, "withTransactionDo" )

	if self._hasActiveTransaction then
		error( "WFSQLiteDatabase:withTransactionDo: nested transaction disallowed", 2 )
	end

	self._nTransactions = self._nTransactions + 1

	self._hasActiveTransaction = true
	self._PERF_queriesInTransaction = {}

	if self._autosaveInterval then
		return withSavepointDo( self, func, ... )
	end

	self:exec "BEGIN TRANSACTION"

	return finish_withTransactionDo( self, nil, pcall( func, ... ) )

end

--------------------------------------------------------------------------------
--[[!
	\brief Temporarily disables foreign key support and applies a function within
		a single database transaction.

	<ul><li>Do not call this from within a function that is already being called by
		\c withTransactionDo (transactions cannot be nested). </li>

	<li>You cannot use \c coroutine.yield within this call.</li></ul>

	\param func The function, which calls functions such as \c :exec to execute
		SQL queries within the transaction, and returns normally to commit.
		If the function returns an error, the transaction is rolled back.

	\param ... Any additional parameters for the function.

	\return The result.
]]--

function databaseMethods:withTransactionAndDisabledForeignKeysDo( func, ... )

	check( 'WFSQLiteDatabase', 'function' )

	logInfo( self, "withTransactionAndDisabledForeignKeysDo" )

	if self._hasActiveTransaction then
		error( "WFSQLiteDatabase:withTransactionDo: nested transaction disallowed", 2 )
	end

	self._nTransactions = self._nTransactions + 1

	self._hasActiveTransaction = true
	self._PERF_queriesInTransaction = {}

	if self._autosaveInterval then
		return withSavepointDo( self, func, ... )
	end
	
	local foreign_keys = self:exec "PRAGMA foreign_keys"
	
	foreign_keys = foreign_keys and foreign_keys.foreign_keys == 1 or false

	if foreign_keys then
	 	self:exec "PRAGMA foreign_keys = OFF"
	end

	self:exec "BEGIN TRANSACTION"

	local function onTransactionFinished()
	
		if foreign_keys then
			self:exec "PRAGMA foreign_keys = ON"
		end
	
	end

	return finish_withTransactionDo( self, onTransactionFinished, pcall( func, ... ) )

end

--------------------------------------------------------------------------------
--[[!
	\brief Requests use of savepoints instead of transactions and sets the
		maximum interval between actual transaction boundaries.

	\param interval (number) number of seconds to allow between transaction
		boundaries (must be >= 1)

	NOTE: You can change the transaction interval on an existing database, but
	the new interval will not take effect until any existing transaction commit
	has completed.

	You can not switch back from autosave to individual transactions on an
	existing database object (at runtime).
]]--

function databaseMethods:setAutosaveInterval( interval )

	check( 'WFSQLiteDatabase', 'number' )

	if interval < 1 then
		error( "WFSQLiteDatabase:setAutosaveInterval: interval must be >= 1 second", 2 )
	end

	logDebug( self, "setAutosaveInterval %.2f", interval )

	self._autosaveInterval = interval

end

--------------------------------------------------------------------------------
--[[!
	\brief Force a transaction commit now.

	This is a no-op if no savepoints have been executed that have not been
	saved.
]]--

function databaseMethods:save()

	check( 'WFSQLiteDatabase' )

	if self._autosaveTimerDisposable then

		self._autosaveTimerDisposable:dispose()
		self._autosaveTimerDisposable = false

		-- Release outer savepoint.

		logDebug( self, "Commit outer savepoint" )

		local start = logger:will_info() and KSDate.timeNow() or nil

		self:exec 'RELEASE outer'

		if logger:will_info() and start then

			local duration = KSDate.timeNow() - start
			local logLevel = duration > 0.5 and "warnf" or "infof"

			logger[ logLevel ]( logger, "Commit outer savepoint took %s seconds for %s",
								duration, self._path )

		end

	else

		logDebug( self, ":save() called, but no savepoint outstanding" )

	end

end

--------------------------------------------------------------------------------
--[[!
	\brief Collects performance data for the connection.
]]--

function databaseMethods:PERF_collectQueryPerformanceStats()

	check( 'WFSQLiteDatabase' )

	self._PERF_gatherStats = true

	--debug.print( "Database for path", self._path, "configured to gather perf data" )

	self._PERF_stmtStats = {}

end

--------------------------------------------------------------------------------

local function appendExplainResults( self, eqpPrefix, sql, output )
	
	if string.match( sql, 'temp.ResultSetVirtualTable_' ) then
		return
	end
	
	local eqpSql = string.format( "%s %s", eqpPrefix, sql )

	local eqpRows = self:gatherArrayOfArray( eqpSql )

	killStatement( self, eqpSql )

	if #eqpRows > 0 then

		local eqpRowsFormatted = {}
		for q = 1, #eqpRows do
			eqpRowsFormatted[ q ] = table.concat( eqpRows[ q ], " | " )
		end

		local eqpSummary = table.concat( eqpRowsFormatted, sqlContinuationIndent )

		output[ #output + 1 ] = string.format( "%s%s\n\n", sqlContinuationIndent, eqpSummary )

	else

		output[ #output + 1 ] = "\n"

	end

end

--------------------------------------------------------------------------------
--[[!
	\brief Generates a pretty-print report on performance data.

	\return (table) The report.
]]--

function databaseMethods:PERF_generateQueryStatsReport()

	check( 'WFSQLiteDatabase' )

	if not self._PERF_gatherStats then return end

	local output = {
		"\nQUERY PROFILE SUMMARY FOR DATABASE ", self._path,
		"\n",
		"\n   Count       Min       Max       Total  SQL",
		"\n--------  --------  --------  ----------  -------------",
		"\n\n",
	}

	local sortedStats = {}
	local n = 0
	
	local totalQueryCount, totalQueryTime = 0,0
	for _, v in pairs( self._PERF_stmtStats ) do
		if not v.sql:match( "^EXPLAIN " ) then
			n = n + 1
			totalQueryCount = totalQueryCount + ( v.count or 0 )
			totalQueryTime = totalQueryTime + ( v.total or 0 )
			sortedStats[ n ] = v
		end
	end
	
	-- add fake row for total time
	table.insert( sortedStats, 1, { 
		count = totalQueryCount,
		min = 0,
		max = 0,
		total = totalQueryTime,
		sql = 'TOTALS'
	} )
	
	table.sort( sortedStats, function( a, b )

		if a.total ~= b.total then
			return a.total > b.total
		else
			return a.sql < b.sql
		end

	end )

	for i = 1, #sortedStats do

		local stat = sortedStats[ i ]

		local sql = stat.sql:gsub( "\n%s+", sqlContinuationIndent )

		output[ #output + 1 ] =
			string.format(
				"%8.0f  %8.3f  %8.3f  %10.3f  %s\n",
					stat.count,
					stat.min,
					stat.max,
					stat.total,
					sql )

		if i <= 10 and not stat.sql:match( "^COMMIT" ) and not ( stat.sql == 'TOTALS' ) then
			appendExplainResults( self, "EXPLAIN QUERY PLAN", stat.sql, output )
		else
			output[ #output + 1 ] = '\n'
		end

	end
	
	return table.concat( output )

end

--------------------------------------------------------------------------------
--[[!
	\brief Reports performance data to the debugger.
]]--

function databaseMethods:PERF_dumpQueryPerformanceStats()

	check( 'WFSQLiteDatabase' )

	if not self._PERF_gatherStats then return end

	debug.print( self:PERF_generateQueryStatsReport() )

	self._PERF_gatherStats = false

end

--------------------------------------------------------------------------------
--[[!
	\brief Reports performance data to the diagnostic log.
]]--

function databaseMethods:sendDiagnosticLogData()

	check( 'WFSQLiteDatabase' )

	if not self._PERF_gatherStats then return end

	-- FIXME [scouten 2012-05-31]: Very hasty translation of console log format
	-- to HTML. Maybe change this to "real" HTML someday?

	local rawReport = self:PERF_generateQueryStatsReport()

	local htmlReport = string.format( "<pre>%s</pre>",
									  KSStringUtils.escapeHtml( rawReport ) )

	local idSafeThreadName = WFCurrentThread.DEBUG_getThreadName()
	idSafeThreadName = idSafeThreadName:replace( ' :: ', '+' )

	WFDiagnosticLog.addSectionToLog {

		id = string.format( 'WFSQLiteDatabase-%s-%s',
							self._path,
							idSafeThreadName ),

		name = string.format( "Database Performance: %s",
							  WFCurrentThread.DEBUG_getThreadName() ),

		parentSection = WFDiagnosticLog.Sections.DB_PERFORMANCE,

		weight = 2000,

		body = htmlReport,

	}

end

--------------------------------------------------------------------------------
--[[!
	\brief Retrieves the path of this database as a display string.

	\return The display string.
]]--

function databaseMethods:__tostring()

	check( 'WFSQLiteDatabase' )

	return string.format( "WFSQLiteDatabase( %s )", self._path )

end

--------------------------------------------------------------------------------
--[[!
	\brief Return the number of SQLite transactions that have been executed
		thus far. Intended for debug / testing use only.
]]--

function databaseMethods:nTransactions()

	check( 'WFSQLiteDatabase' )

	return self._nTransactions

end

--------------------------------------------------------------------------------
--[[!
	\brief backs up self into destDB. destDB will be completely clobbered.
]]--

function databaseMethods:onlineBackup( destDB, progressFn )

	check( 'WFSQLiteDatabase', 'WFSQLiteDatabase' )

	self._db:onlineBackup( destDB._db, progressFn )

	return destDB

end

--------------------------------------------------------------------------------
--[[!
	\brief backs up self into destDB. destDB will be completely clobbered
		and closed upon completion or failure.

	\param path (string) filename where the new db should be created.

	\param canceledHandler (function) function to call if error or subscription
		is disposed before completion.

	\return <code>Observable<number> stream of values relaying progress of
		the backup as a fraction indicating percent complete.
]]--

function databaseMethods:onlineBackupWithProgressStream( destDB, canceledOrErrHandler )

	check( 'WFSQLiteDatabase', 'WFSQLiteDatabase' )

	return Rx.Observable.create( function( observer )

		local disposed, completed

		local function relayProgress( pctComplete )
			observer:onNext( pctComplete )
			return not disposed
		end

		local scheduledAction = WFRxScheduler.sharedScheduler():schedule( function()

			local success, errMsg = pcall( self.onlineBackup, self, destDB, relayProgress )

			-- self.onlineBackup has completed no need to schedule the canceledHandler.
			if success then
				completed = true
				observer:onNext( 1 ) -- ensure progress goes to 100%
				observer:onCompleted()
			else
				canceledOrErrHandler()
				canceledOrErrHandler = nil
				observer:onError( errMsg )
			end

		end )

		return Rx.Disposable.create( function()

				disposed = true
				scheduledAction:dispose() -- don't run action if already disposed

				if not completed and canceledOrErrHandler then

					-- Schedule the caneledHandler to prevent the handler
					-- from closing the database while it is still in the
					-- onlineBackup call which would result in bad memory
					-- access error.

					WFRxScheduler.sharedScheduler():schedule( function()
						if canceledOrErrHandler then
							canceledOrErrHandler()
							canceledOrErrHandler = nil
						end
					end )

				end

			end )

	end )

end

--------------------------------------------------------------------------------

--[[!
	\brief Get a stream that fires with SQLite error codes when a database operation
		encounters a critical error. Possible error codes include:

		10 : Some kind of disk I/O error occurred
		11 : The database disk image is malformed
		13 : Insertion failed because database is full

	\return <code>Observable<number>
]]--

function databaseMethods:getSQLiteFatalErrorCodeStream()

	check( 'WFSQLiteDatabase' )

	return self._sqliteFatalErrorCodeStream

		:filter( function( hasError )
			return hasError
		end )

end


--------------------------------------------------------------------------------

return Rx.Class 'WFSQLiteDatabase'( databaseMethods )
