import { FinancialBalance, User } from "../db/interfaces.ts"; import { sqlCreateUserRole, sqlDeleteSessionConfig, sqlDeleteStateKey, sqlDeleteUser, sqlDeleteUserPin, sqlDeleteUserRole, sqlDeleteUserTokens, sqlGetAllFinancialBalances, sqlGetAllRoles, sqlGetAllSessionConfigs, sqlGetAllStateKeys, sqlGetAllUserPins, sqlGetAllUserRoles, sqlGetAllUsers, sqlGetAllUserTokens, sqlGetFinancialBalancesByUser, sqlGetSessionConfig, sqlGetStateKeys, sqlGetUser, sqlGetUserPin, sqlGetUserRoles, sqlGetUserTokens, sqlSetSessionConfig, sqlSetStateKey, sqlSetUserPin, sqlSetUserTokens, sqlUpdateSessionConfig as _sqlUpdateSessionConfig, sqlUpdateUserPin as _sqlUpdateUserPin, sqlUpsertFinancialBalance, sqlUpsertUser as _sqlUpsertUser, sqlUpsertUserFull, } from "../db/sqlUserRepository.ts"; export function handleDatabaseError(error: unknown): void { const errorMessage = error instanceof Error ? error.message : String(error); console.error( "🔄 Database error detected. App will continue in degraded mode...", ); console.error("Error details:", errorMessage); // Don't restart the app - just continue with degraded functionality } interface KvAdapter { get(key: Deno.KvKey): Promise<{ value: T | null }>; set(key: Deno.KvKey, value: T): Promise; delete(key: Deno.KvKey): Promise; list( options?: { prefix?: Deno.KvKey }, ): AsyncIterable<{ key: Deno.KvKey; value: T }>; close(): Promise | void; } export class KvCacheStore { private memKv: KvAdapter; private fileKv: KvAdapter; constructor(memKv: KvAdapter, fileKv: KvAdapter) { this.memKv = memKv; this.fileKv = fileKv; } static async initialize(filePath: string): Promise { try { await Deno.mkdir(filePath, { recursive: true }); } catch (err) { if (!(err instanceof Deno.errors.AlreadyExists)) { console.error("Failed to create KV cache directory:", err); } } const memKv: KvAdapter = await Deno.openKv(":memory:") as KvAdapter; const fileKv = await Deno.openKv(`${filePath}/kv-cache`) as KvAdapter; for await (const entry of fileKv.list({ prefix: [] })) { await memKv.set(entry.key, entry.value); } return new KvCacheStore(memKv, fileKv); } private logKvError( operation: string, error: unknown, context?: Record, ): void { const message = error instanceof Error ? error.message : String(error); console.error(`[KvCacheStore] ${operation} failed`, { ...context, error: message, }); if (error instanceof Error && error.stack) { console.error(error.stack); } } async get(key: Deno.KvKey): Promise { try { const result = await this.memKv.get(key); if (result.value !== null) { return result.value; } const sqlResult = await this.fetchFromSql(key); if (sqlResult) { await this.memKv.set(key, sqlResult); await this.fileKv.set(key, sqlResult); return sqlResult; } return null; } catch (error) { this.logKvError("get", error, { key: JSON.stringify(key) }); handleDatabaseError(error); return null; } } async set(key: Deno.KvKey, value: T): Promise { try { const existing = await this.get(key); if ( existing && typeof existing === "object" && existing !== null && "created_on" in existing ) { const createdOn = (existing as unknown as { created_on: number }).created_on; if (createdOn) { value = { ...value, created_on: createdOn, }; } } await this.memKv.set(key, value); await this.fileKv.set(key, value); await this.persistToSql(key, value); } catch (error) { console.error("KV/SQL set failed:", error); this.logKvError("set", error, { key: JSON.stringify(key) }); handleDatabaseError(error); throw error instanceof Error ? error : new Error(String(error)); } } async delete(key: Deno.KvKey): Promise { try { await this.memKv.delete(key); await this.fileKv.delete(key); await this.deleteFromSql(key); } catch (error) { this.logKvError("delete", error, { key: JSON.stringify(key) }); handleDatabaseError(error); } } async *list( options?: { prefix?: Deno.KvKey }, ): AsyncIterable<{ key: Deno.KvKey; value: T }> { try { const yieldedKeys = new Set(); for await (const entry of this.memKv.list(options ?? {})) { const keyStr = JSON.stringify(entry.key); if (!yieldedKeys.has(keyStr)) { yieldedKeys.add(keyStr); yield entry; } } for await (const entry of this.fileKv.list(options ?? {})) { const keyStr = JSON.stringify(entry.key); if (!yieldedKeys.has(keyStr)) { yieldedKeys.add(keyStr); yield entry; } } const prefix = options?.prefix; const namespace = prefix && prefix.length > 0 ? String(prefix[0]) : null; if (namespace) { const sqlEntries = await this.listFromSql(namespace); for (const entry of sqlEntries) { const keyStr = JSON.stringify(entry.key); if (!yieldedKeys.has(keyStr)) { if (prefix && !this.keyMatchesPrefix(entry.key, prefix)) { continue; } yieldedKeys.add(keyStr); yield entry; } } } } catch (error) { this.logKvError("list", error, { prefix: JSON.stringify(options?.prefix ?? []), }); handleDatabaseError(error); throw error instanceof Error ? error : new Error(String(error)); } } private keyMatchesPrefix(key: Deno.KvKey, prefix: Deno.KvKey): boolean { if (key.length < prefix.length) { return false; } for (let i = 0; i < prefix.length; i++) { if (key[i] !== prefix[i]) { return false; } } return true; } private async listFromSql( namespace: string, ): Promise> { const results: Array<{ key: Deno.KvKey; value: T }> = []; switch (namespace) { case "users": { const users = await sqlGetAllUsers(); for (const user of users) { results.push({ key: ["users", user.id], value: { id: user.id, name: user.name || "", email: user.email || "", phone: user.phone || "", address_1: user.address_1 || "", address_2: user.address_2 || "", city: user.city || "", state: user.state || "", zip: user.zip || "", country: user.country || "", user_type: user.user_type, onboarded_landlord: user.onboarded_landlord ?? false, onboarded_tenant: user.onboarded_tenant ?? false, email_verified: user.email_verified, banking_setup_complete: user.banking_setup_complete, onboarding_step: user.onboarding_step, encrypted_data: user.encrypted_data, started_on: user.started_on || Date.now(), updated_on: user.updated_on || Date.now(), } as T, }); } break; } case "roles": { const roles = await sqlGetAllRoles(); for (const role of roles) { results.push({ key: ["roles", role.id], value: { id: role.id, description: role.description || "", } as T, }); } break; } case "user_roles": { const userRoles = await sqlGetAllUserRoles(); const grouped = new Map(); for (const ur of userRoles) { const existing = grouped.get(ur.userId || "") || []; if (!ur.userId) { continue; } existing.push(ur.roleId || ""); grouped.set(ur.userId || "", existing); } for (const [userId, roleIds] of grouped) { results.push({ key: ["user_roles", userId], value: roleIds as T, }); } break; } case "state_keys": { const stateKeys = await sqlGetAllStateKeys(); for (const sk of stateKeys) { results.push({ key: ["state_keys", sk.state_id], value: { seal_key: sk.seal_key, state_id: sk.state_id, cookie_hash: sk.cookie_hash, created_on: sk.created_on, } as T, }); } break; } case "financial_balances": { const balances = await sqlGetAllFinancialBalances(); for (const balance of balances) { results.push({ key: ["financial_balances", balance.user_id || ""], value: balance as T, }); } break; } case "user_tokens": { const tokens = await sqlGetAllUserTokens(); for (const token of tokens) { results.push({ key: ["user_tokens", token.user_id || ""], value: { user_id: token.user_id, state_id: token.state_id, access_token: token.access_token, refresh_token: token.refresh_token, created_on: token.created_on, updated_on: token.updated_on, } as T, }); } break; } case "user_pins": { const pins = await sqlGetAllUserPins(); for (const pin of pins) { results.push({ key: ["user_pins", pin.user_id || ""], value: { user_id: pin.user_id, pin: pin.pin, created_on: pin.created_on, modified_on: pin.modified_on, } as T, }); } break; } case "session_configs": { const configs = await sqlGetAllSessionConfigs(); for (const config of configs) { results.push({ key: [ "session_configs", config.user_id || "", config.device_id || "", ], value: { device_id: config.device_id, user_id: config.user_id, other_data: config.other_data, active_date: config.active_date, active_route: config.active_route, page_data: config.page_data, } as T, }); } break; } default: break; } return results; } async close() { try { await this.memKv.close(); await this.fileKv.close(); } catch (error) { this.logKvError("close", error); handleDatabaseError(error); throw error instanceof Error ? error : new Error(String(error)); } } private async persistToSql(key: Deno.KvKey, value: unknown): Promise { const [namespace, id] = key; if (typeof namespace !== "string") { return; } switch (namespace) { case "users": { const user = value as User; await sqlUpsertUserFull({ id: user.id, name: user.name ?? "", email: user.email ?? "", phone: user.phone ?? "", address_1: user.address_1 ?? "", address_2: user.address_2 ?? "", city: user.city ?? "", state: user.state ?? "", zip: user.zip ?? "", country: user.country ?? "", user_type: user.user_type ?? null, onboarded_landlord: user.onboarded_landlord ?? false, onboarded_tenant: user.onboarded_tenant ?? false, email_verified: user.email_verified ?? null, banking_setup_complete: user.banking_setup_complete ?? null, onboarding_step: user.onboarding_step ?? null, notification_preferences_json: user.notification_preferences_json ?? null, payment_preferences_json: user.payment_preferences_json ?? null, encrypted_data: user.encrypted_data ?? null, started_on: user.started_on ?? Date.now(), updated_on: user.updated_on ?? Date.now(), }); break; } case "roles": { const role = value as { id: string; description: string }; void role; break; } case "user_roles": { const roleIds = value as string[]; const userId = id as string; await sqlDeleteUserRole(userId); for (const roleId of roleIds) { await sqlCreateUserRole(userId, roleId); } break; } case "state_keys": { const stateKey = value as { seal_key: string; state_id: string; cookie_hash: string; created_on: number; }; await sqlSetStateKey( stateKey.seal_key, stateKey.state_id, stateKey.cookie_hash, ); break; } case "financial_balances": { const balance = value as FinancialBalance; await sqlUpsertFinancialBalance( balance.user_id || "", balance.balance || 0, balance.balance_date || 0, ); break; } case "user_tokens": { const token = value as { user_id: string; state_id: string; access_token: string; refresh_token: string; }; await sqlSetUserTokens( token.user_id, token.state_id, token.access_token, token.refresh_token, ); break; } case "user_pins": { const userPin = value as { user_id: string; pin: string; created_on: number; modified_on: number; }; await sqlSetUserPin(userPin.user_id, userPin.pin); break; } case "session_configs": { const sessionConfig = value as { device_id: string; user_id: string; other_data: string; active_date: number; active_route: string; page_data: string; }; await sqlSetSessionConfig( sessionConfig.device_id, sessionConfig.user_id, sessionConfig.other_data, sessionConfig.active_route, sessionConfig.page_data, ); break; } default: break; } } private async deleteFromSql(key: Deno.KvKey): Promise { const [namespace, id] = key; if (typeof namespace !== "string") { return; } switch (namespace) { case "users": await sqlDeleteUser(id as string); break; case "user_roles": await sqlDeleteUserRole(id as string); break; case "state_keys": await sqlDeleteStateKey(id as string); break; case "user_tokens": await sqlDeleteUserTokens(id as string); break; case "user_pins": await sqlDeleteUserPin(id as string); break; case "session_configs": { const recordId = key[2] as string | undefined; if (recordId) await sqlDeleteSessionConfig(recordId); break; } default: break; } } private async fetchFromSql(key: Deno.KvKey): Promise { const [namespace, id] = key; if (typeof namespace !== "string") { return null; } switch (namespace) { case "users": { const user = await sqlGetUser(id as string); if (!user) return null; return { id: user.id, name: user.name || "", email: user.email || "", phone: user.phone || "", address_1: user.address_1 || "", address_2: user.address_2 || "", city: user.city || "", state: user.state || "", zip: user.zip || "", country: user.country || "", user_type: user.user_type, onboarded_landlord: user.onboarded_landlord ?? false, onboarded_tenant: user.onboarded_tenant ?? false, email_verified: user.email_verified, banking_setup_complete: user.banking_setup_complete, onboarding_step: user.onboarding_step, encrypted_data: user.encrypted_data, started_on: user.started_on || Date.now(), updated_on: user.updated_on || Date.now(), } as T; } case "roles": { const roles = await sqlGetAllRoles(); const role = roles.find(( r: { id: string; description: string | null }, ) => r.id === id); if (!role) return null; return { id: role.id, description: role.description || "", } as T; } case "user_roles": { const roles = await sqlGetUserRoles(id as string); if (!roles || roles.length === 0) return null; return roles.map((r: { roleId: string | null }) => r.roleId).filter(( roleId: string | null, ) => roleId !== "") as T; } case "state_keys": { const stateKeys = await sqlGetStateKeys(id as string); if (!stateKeys || stateKeys.length === 0) return null; const sortedKeys = stateKeys.sort(( a: { created_on: number }, b: { created_on: number }, ) => b.created_on - a.created_on); return { seal_key: sortedKeys[0].seal_key, state_id: sortedKeys[0].state_id, cookie_hash: sortedKeys[0].cookie_hash, created_on: sortedKeys[0].created_on, } as T; } case "financial_balances": { const balances = await sqlGetFinancialBalancesByUser(id as string); if (!balances || balances.length === 0) return null; return balances[0] as T; } case "user_tokens": { const tokens = await sqlGetUserTokens(id as string); if (!tokens || tokens.length === 0) return null; const token = tokens[0]; // Get the first token return { user_id: token.user_id, state_id: token.state_id, access_token: token.access_token, refresh_token: token.refresh_token, created_on: token.created_on, updated_on: token.updated_on, } as T; } case "user_pins": { const userPin = await sqlGetUserPin(id as string); if (!userPin) return null; return { user_id: userPin.user_id, pin: userPin.pin, created_on: userPin.created_on, modified_on: userPin.modified_on, } as T; } case "session_configs": { const owner = key[1] as string | undefined; if (!owner) return null; const sessionConfig = await sqlGetSessionConfig(owner); if (!sessionConfig) return null; return { device_id: sessionConfig.device_id, user_id: sessionConfig.user_id, other_data: sessionConfig.other_data, active_date: sessionConfig.active_date, active_route: sessionConfig.active_route, page_data: sessionConfig.page_data, } as T; } default: return null; } } } const defaultPersistDir = "./data"; const persistPath = Deno.env.get("KV_CACHE_PATH") ?? defaultPersistDir; export const kvStore = await KvCacheStore.initialize(persistPath);