Spaces:

HonzaH
/

Letschat

Sleeping

Letschat / src /lib /persistence.ts
Upload 192 files
cc015e6 verified 11 months ago
30 kB
import { createServerClient, createServiceClient } from 'lib/supabase/server'
import { randomUUID } from 'crypto'
import type { Database } from 'types/database'
import type { SupabaseClient } from '@supabase/supabase-js'
import { Pool } from 'pg'
import { promises as fs } from 'fs'
import path from 'path'
import { logEvent } from './logger'
import { allowInMemoryFallback } from './supabase/env'
import { isSupabaseConfigured } from './supabase/env'

const LOCAL_DB_URL = process.env.LOCAL_DATABASE_URL
const FILE_STORE_PATH = process.env.LOCAL_FILE_STORE_PATH || path.join(process.cwd(), '.data', 'store.json')
const ENABLE_FILE_STORE = process.env.LOCAL_FILE_STORE === '1'

let pool: Pool | null = null
if (LOCAL_DB_URL) {
  pool = new Pool({ connectionString: LOCAL_DB_URL })
}

// In-memory fallback for local/dev testing when DB/Supabase is unavailable
const inMemory = {
  codes: [] as Array<Record<string, any>>,
  rooms: [] as Array<Record<string, any>>,
  room_messages: [] as Array<Record<string, any>>,
  logs: [] as Array<Record<string, any>>
}

function isDevFallbackAllowed() {
  // Allow in-memory fallback during local development when NODE_ENV !== 'production'.
  if (allowInMemoryFallback) return true
  return process.env.NODE_ENV !== 'production'
}

type FileStore = {
  codes: Array<Record<string, any>>
  rooms: Array<Record<string, any>>
  room_messages: Array<Record<string, any>>
}

async function ensureFileStoreDir() {
  const dir = path.dirname(FILE_STORE_PATH)
  await fs.mkdir(dir, { recursive: true })
}

async function readFileStore(): Promise<FileStore> {
  if (!ENABLE_FILE_STORE) return { codes: [], rooms: [], room_messages: [] }
  try {
    const buf = await fs.readFile(FILE_STORE_PATH, 'utf8')
    return JSON.parse(buf) as FileStore
  } catch {
    return { codes: [], rooms: [], room_messages: [] }
  }
}

async function writeFileStore(store: FileStore) {
  if (!ENABLE_FILE_STORE) return
  await ensureFileStoreDir()
  await fs.writeFile(FILE_STORE_PATH, JSON.stringify(store, null, 2), 'utf8')
}

async function getSupabase(): Promise<SupabaseClient<Database>> {
  return createServerClient()
}

// Prefer service-role client for write operations to avoid RLS blocks when running on Supabase.
async function getSupabaseWrite(): Promise<SupabaseClient<Database>> {
  if (process.env.SUPABASE_SERVICE_ROLE_KEY) {
    return createServiceClient()
  }
  return createServerClient()
}

function pickFallbackStore() {
  // Prefer file store when enabled; otherwise use in-memory store
  return ENABLE_FILE_STORE ? readFileStore() : Promise.resolve({ codes: inMemory.codes, rooms: inMemory.rooms, room_messages: inMemory.room_messages })
}

// Codes

// Helper to fetch a code by id and its linked counterpart (if any).
// Returns primary (the requested code) and linked (the paired code when present).
export async function getCodeWithLinked(codeId: string) {
  const primary = await getCodeById(codeId)
  if (!primary) return { primary: null, linked: null }

  // If linked_to is set, fetch that code; otherwise try to find a code that links to this one.
  let linked = primary.linked_to ? await getCodeById(primary.linked_to) : null
  if (!linked) {
    if (pool) {
      const res = await pool.query('SELECT * FROM codes WHERE linked_to = $1 LIMIT 1', [codeId])
      linked = res.rows[0] || null
    } else {
      try {
        const supabase = await getSupabase()
        const { data } = await supabase.from('codes').select('*').eq('linked_to', codeId).limit(1)
        linked = data && data.length > 0 ? (data[0] as any) : null
      } catch (e) {
        console.error('[persistence.getCodeWithLinked] supabase error (reverse lookup):', e)
        if (isDevFallbackAllowed()) {
          linked = inMemory.codes.find((c) => c.linked_to === codeId) || null
        }
      }
    }
  }

  return { primary, linked }
}

export async function getCodeById(codeId: string) {
  if (pool) {
    const res = await pool.query('SELECT * FROM codes WHERE id = $1 LIMIT 1', [codeId])
    return res.rows[0] || null
  }
  try {
    const supabase = await getSupabase()
    const { data } = await supabase.from('codes').select('*').eq('id', codeId).single()
    return data || null
  } catch (e) {
    console.error('[persistence.getCodeById] supabase error:', e)
    if (isDevFallbackAllowed()) {
      return inMemory.codes.find((c) => c.id === codeId) || null
    }
    return null
  }
}

export async function getCodeByCode(code: string) {
  if (pool) {
    const res = await pool.query('SELECT * FROM codes WHERE code = $1 LIMIT 1', [code.toUpperCase()])
    return res.rows[0] || null
  }
  try {
    const supabase = await getSupabase()
    const { data } = await supabase.from('codes').select('*').eq('code', code.toUpperCase()).single()
    return data || null
  } catch (e) {
    console.error('[persistence.getCodeByCode] supabase error:', e)
    if (isDevFallbackAllowed()) {
      return inMemory.codes.find((c) => c.code === code.toUpperCase()) || null
    }
    return null
  }
}

export async function getCodeForUserInRoom(roomId: string, userId: string) {
  if (pool) {
    const res = await pool.query('SELECT id FROM codes WHERE room_id = $1 AND user_id = $2 LIMIT 1', [roomId, userId])
    return res.rows[0] || null
  }
  try {
    const supabase = await getSupabaseWrite()
    const { data } = await supabase.from('codes').select('id').eq('room_id', roomId).eq('user_id', userId).limit(1)
    return data && data.length > 0 ? data[0] : null
  } catch (e) {
    console.error('[persistence.getCodeForUserInRoom] supabase error:', e)
    if (isDevFallbackAllowed()) {
      return inMemory.codes.find((c) => c.room_id === roomId && c.user_id === userId) || null
    }
    return null
  }
}

export async function getAnyCodeForRoom(roomId: string) {
  if (pool) {
    const res = await pool.query('SELECT id FROM codes WHERE room_id = $1 LIMIT 1', [roomId])
    return res.rows[0] || null
  }
  try {
    const supabase = await getSupabaseWrite()
    const { data } = await supabase.from('codes').select('id').eq('room_id', roomId).limit(1)
    return data && data.length > 0 ? data[0] : null
  } catch (e) {
    console.error('[persistence.getAnyCodeForRoom] supabase error:', e)
    if (isDevFallbackAllowed()) {
      return inMemory.codes.find((c) => c.room_id === roomId) || null
    }
    return null
  }
}

export async function insertCode(payload: Database['public']['Tables']['codes']['Insert']) {
  const normalizedPayload: Database['public']['Tables']['codes']['Insert'] = {
    ...payload,
    code: payload.code.toUpperCase()
  }

  if (pool) {
    const cols = Object.keys(normalizedPayload)
    const vals = Object.values(normalizedPayload)
    const idx = vals.map((_, i) => `$${i + 1}`).join(',')
    const query = `INSERT INTO codes(${cols.join(',')}) VALUES(${idx}) RETURNING *`
    const res = await pool.query(query, vals)
    await logEvent({ module: 'codes', operation: 'insert', data: res.rows[0] })
    return res.rows[0]
  }
  const supabase = await getSupabaseWrite()
  try {
    const { data, error } = await supabase.from('codes').insert(normalizedPayload).select().single()
    if (error) {
      console.error('[persistence.insertCode] anon insert error:', error)
      // Try service client if available
      try {
        if (process.env.SUPABASE_SERVICE_ROLE_KEY) {
          const service = createServiceClient()
          const { data: sdata, error: serror } = await service.from('codes').insert(normalizedPayload).select().single()
          if (serror) {
            console.error('[persistence.insertCode] service insert error:', serror)
            // fall through to dev fallback if allowed
          } else {
            const row = sdata || null
            await logEvent({ module: 'codes', operation: 'insert', data: row })
            return row
          }
        }
      } catch (svcErr) {
        console.error('[persistence.insertCode] service client attempt failed:', svcErr)
      }
      // If we reach here, supabase insert(s) failed. Use in-memory fallback in dev.
      if (isDevFallbackAllowed()) {
        const id = randomUUID()
        const now = new Date().toISOString()
        const row = {
          id,
          code: normalizedPayload.code,
          linked_to: normalizedPayload.linked_to || null,
          used: normalizedPayload.used ?? null,
          user_id: normalizedPayload.user_id || null,
          room_id: normalizedPayload.room_id || null,
          pin_hash: normalizedPayload.pin_hash || null,
          session_hash: normalizedPayload.session_hash || null,
          date_first: normalizedPayload.date_first || null,
          date_last: normalizedPayload.date_last || null,
          used_count: normalizedPayload.used_count ?? 0,
          created_at: normalizedPayload.created_at || now
        }
        inMemory.codes.push(row)
        console.warn('[persistence.insertCode] falling back to in-memory storage (dev only)')
        await logEvent({ module: 'codes', operation: 'insert_fallback', data: row, error: error?.message || null })
        return row
      }
      return null
    }
    const row = data || null
    await logEvent({ module: 'codes', operation: 'insert', data: row })
    return row
  } catch (e) {
    console.error('[persistence.insertCode] unexpected error:', e)
    if (isDevFallbackAllowed()) {
      const id = randomUUID()
      const now = new Date().toISOString()
      const row = {
        id,
        code: normalizedPayload.code,
        linked_to: normalizedPayload.linked_to || null,
        used: normalizedPayload.used ?? null,
        user_id: normalizedPayload.user_id || null,
        room_id: normalizedPayload.room_id || null,
        pin_hash: normalizedPayload.pin_hash || null,
        session_hash: normalizedPayload.session_hash || null,
        date_first: normalizedPayload.date_first || null,
        date_last: normalizedPayload.date_last || null,
        used_count: normalizedPayload.used_count ?? 0,
        created_at: normalizedPayload.created_at || now
      }
      inMemory.codes.push(row)
      console.warn('[persistence.insertCode] unexpected error - falling back to in-memory (dev only)')
      await logEvent({ module: 'codes', operation: 'insert_fallback', data: row, error: (e as Error)?.message || String(e) })
      return row
    }
    return null
  }
}

export async function updateCode(id: string, payload: Partial<Database['public']['Tables']['codes']['Update']>) {
  if (pool) {
    const cols = Object.keys(payload)
    const vals = Object.values(payload)
    const set = cols.map((c, i) => `${c} = $${i + 1}`).join(',')
    const query = `UPDATE codes SET ${set} WHERE id = $${cols.length + 1} RETURNING *`
    const res = await pool.query(query, [...vals, id])
    await logEvent({ module: 'codes', operation: 'update', data: res.rows[0] })
    return res.rows[0]
  }
  const supabase = await getSupabaseWrite()
  try {
    const { data, error } = await supabase.from('codes').update(payload).eq('id', id).select().single()
    if (error) {
      console.error('[persistence.updateCode] supabase update error:', error)
      if (isDevFallbackAllowed()) {
        const idx = inMemory.codes.findIndex((c) => c.id === id)
        if (idx >= 0) {
          inMemory.codes[idx] = { ...inMemory.codes[idx], ...payload }
          await logEvent({ module: 'codes', operation: 'update_fallback', data: inMemory.codes[idx], error: error.message || null })
          return inMemory.codes[idx]
        }
      }
      return null
    }
    const row = data || null
    await logEvent({ module: 'codes', operation: 'update', data: row })
    return row
  } catch (e) {
    console.error('[persistence.updateCode] supabase error (exception):', e)
    if (isDevFallbackAllowed()) {
      const idx = inMemory.codes.findIndex((c) => c.id === id)
      if (idx >= 0) {
        inMemory.codes[idx] = { ...inMemory.codes[idx], ...payload }
        await logEvent({ module: 'codes', operation: 'update_fallback', data: inMemory.codes[idx], error: (e as Error)?.message || String(e) })
        return inMemory.codes[idx]
      }
    }
    return null
  }
}

// Rooms
export async function insertRoom(payload: Partial<Database['public']['Tables']['rooms']['Insert']>) {
  if (pool) {
    const cols = Object.keys(payload)
    const vals = Object.values(payload)
    const idx = vals.map((_, i) => `$${i + 1}`).join(',')
    const query = `INSERT INTO rooms(${cols.join(',')}) VALUES(${idx}) RETURNING *`
    const res = await pool.query(query, vals)
    await logEvent({ module: 'rooms', operation: 'insert', data: res.rows[0] })
    return res.rows[0]
  }
  try {
    const supabase = await getSupabaseWrite()
    const { data, error } = await supabase.from('rooms').insert(payload).select().single()
    if (error) {
      console.error('[persistence.insertRoom] supabase error:', error)
      if (isDevFallbackAllowed()) {
        const id = randomUUID()
        const now = new Date().toISOString()
        const row = {
          id,
          name: payload.name || null,
          description: payload.description || null,
          status: payload.status ?? 1,
          date_created: payload.date_created || now,
          date_last_message: payload.date_last_message || null,
          updated_at: payload.updated_at || now
        }
        inMemory.rooms.push(row)
        console.warn('[persistence.insertRoom] falling back to in-memory storage (dev only)')
        await logEvent({ module: 'rooms', operation: 'insert_fallback', data: row, error: error.message || null })
        return row
      }
      return null
    }
    await logEvent({ module: 'rooms', operation: 'insert', data: data || null })
    return data || null
  } catch (e) {
    console.error('[persistence.insertRoom] supabase error:', e)
    if (isDevFallbackAllowed()) {
      const id = randomUUID()
      const now = new Date().toISOString()
      const row = {
        id,
        name: payload.name || null,
        description: payload.description || null,
        status: payload.status ?? 1,
        date_created: payload.date_created || now,
        date_last_message: payload.date_last_message || null,
        updated_at: payload.updated_at || now
      }
      inMemory.rooms.push(row)
      console.warn('[persistence.insertRoom] falling back to in-memory storage (dev only)')
      await logEvent({ module: 'rooms', operation: 'insert_fallback', data: row, error: (e as Error)?.message || String(e) })
      return row
    }
    return null
  }
}

export async function getRoomById(roomId: string) {
  if (pool) {
    try {
      const res = await pool.query('SELECT * FROM rooms WHERE id = $1 LIMIT 1', [roomId])
      console.log('[persistence.getRoomById] using pool, found=', !!res.rows[0])
      return res.rows[0] || null
    } catch (e) {
      console.error('[persistence.getRoomById] pool error:', e)
    }
  }
  try {
    console.log('[persistence.getRoomById] no pool, isSupabaseConfigured=', isSupabaseConfigured())
    const supabase = await getSupabaseWrite()
    const { data } = await supabase.from('rooms').select('*').eq('id', roomId).single()
    console.log('[persistence.getRoomById] supabase returned=', !!data)
    return data || null
  } catch (e) {
    console.error('[persistence.getRoomById] supabase error:', e)
    if (isDevFallbackAllowed()) {
      const found = inMemory.rooms.find((r) => r.id === roomId) || null
      console.log('[persistence.getRoomById] falling back to inMemory, found=', !!found)
      return found
    }
    return null
  }
}

export async function updateRoom(roomId: string, payload: Partial<Database['public']['Tables']['rooms']['Update']>) {
  if (pool) {
    const cols = Object.keys(payload)
    const vals = Object.values(payload)
    const set = cols.map((c, i) => `${c} = $${i + 1}`).join(',')
    const query = `UPDATE rooms SET ${set} WHERE id = $${cols.length + 1} RETURNING *`
    const res = await pool.query(query, [...vals, roomId])
    await logEvent({ module: 'rooms', operation: 'update', data: res.rows[0] })
    return res.rows[0]
  }
  try {
    const supabase = await getSupabaseWrite()
    const { data, error } = await supabase.from('rooms').update(payload).eq('id', roomId).select().single()
    if (error) {
      console.error('[persistence.updateRoom] supabase error:', error)
      if (isDevFallbackAllowed()) {
        const idx = inMemory.rooms.findIndex((r) => r.id === roomId)
        if (idx >= 0) {
          inMemory.rooms[idx] = { ...inMemory.rooms[idx], ...payload }
          await logEvent({ module: 'rooms', operation: 'update_fallback', data: inMemory.rooms[idx], error: error.message || null })
          return inMemory.rooms[idx]
        }
      }
      return null
    }
    const row = data || null
    await logEvent({ module: 'rooms', operation: 'update', data: row })
    return row
  } catch (e) {
    console.error('[persistence.updateRoom] supabase error (exception):', e)
    if (isDevFallbackAllowed()) {
      const idx = inMemory.rooms.findIndex((r) => r.id === roomId)
      if (idx >= 0) {
        inMemory.rooms[idx] = { ...inMemory.rooms[idx], ...payload }
        await logEvent({ module: 'rooms', operation: 'update_fallback', data: inMemory.rooms[idx], error: (e as Error)?.message || String(e) })
        return inMemory.rooms[idx]
      }
    }
    return null
  }
}

export async function updateCodesRoomId(codeIds: string[], roomId: string) {
  if (pool) {
    const query = `UPDATE codes SET room_id = $1 WHERE id = ANY($2::uuid[]) RETURNING id, room_id`
    const res = await pool.query(query, [roomId, codeIds])
    await logEvent({ module: 'codes', operation: 'update_room', data: res.rows })
    return res.rows
  }
  try {
    const supabase = await getSupabase()
    const { data } = await supabase.from('codes').update({ room_id: roomId }).in('id', codeIds).select('id, room_id')
    return data || []
  } catch (e) {
    console.error('[persistence.updateCodesRoomId] supabase error:', e)
    if (isDevFallbackAllowed()) {
      const updated: Array<{ id: string; room_id: string } > = []
      for (const id of codeIds) {
        const idx = inMemory.codes.findIndex((c) => c.id === id)
        if (idx >= 0) {
          inMemory.codes[idx].room_id = roomId
          updated.push({ id, room_id: roomId })
        }
      }
      await logEvent({ module: 'codes', operation: 'update_room_fallback', data: updated, error: (e as Error)?.message || String(e) })
      return updated
    }
    return []
  }
}

// Rooms listing for dashboard
type RoomWithCodes = {
  id: string
  name?: string | null
  description?: string | null
  status?: number | null
  date_created: string
  date_last_message?: string | null
  message_count: number
  codes: Array<{ id: string; code: string }>
}

export async function getRoomsForUser(userId: string): Promise<RoomWithCodes[]> {
  if (!userId) return []

  // Helper to aggregate rows into unique rooms with codes
  const aggregate = (rows: Array<Record<string, any>>): RoomWithCodes[] => {
    const map = new Map<string, RoomWithCodes>()
    for (const row of rows) {
      const roomId = row.room_id || row.id
      if (!roomId) continue
      const existing: RoomWithCodes = map.get(roomId) || {
        id: roomId,
        name: row.name ?? null,
        description: row.description ?? null,
        status: row.status ?? 1,
        date_created: row.date_created || row.created_at || new Date().toISOString(),
        date_last_message: row.date_last_message || row.updated_at || null,
        message_count: Number(row.message_count ?? 0) || 0,
        codes: []
      }
      if (row.code_id && row.code) {
        existing.codes.push({ id: row.code_id, code: row.code })
      } else if (row.id && row.code) {
        existing.codes.push({ id: row.id, code: row.code })
      }
      map.set(roomId, existing)
    }
    return Array.from(map.values())
  }

  // 1) Postgres (preferred)
  if (pool) {
    const res = await pool.query(
      `
      SELECT
        r.*,
        COALESCE(r.date_last_message, r.updated_at, r.date_created) AS date_last_message,
        (SELECT COUNT(*) FROM room_messages m WHERE m.room_id = r.id) AS message_count,
        c.id AS code_id,
        c.code
      FROM rooms r
      LEFT JOIN codes c ON c.room_id = r.id
      WHERE (c.user_id = $1 OR r.creator_id = $1) AND r.status = 1
      ORDER BY COALESCE(r.date_last_message, r.updated_at, r.date_created) DESC
      `,
      [userId]
    )
    return aggregate(res.rows)
  }

  // 2) Supabase
  if (isSupabaseConfigured()) {
    try {
      const supabase = await getSupabaseWrite()

      // 1) Fetch rooms where user is creator (primary source)
      let rows: Array<Record<string, any>> = []
      try {
        const { data: creatorRooms, error: creatorErr } = await supabase
          .from('rooms')
          .select('*, room_messages(count)')
          .eq('creator_id', userId)
          .eq('status', 1)
        if (creatorErr) {
          console.error('[persistence.getRoomsForUser] supabase creator rooms error:', creatorErr)
        } else if (creatorRooms && Array.isArray(creatorRooms) && creatorRooms.length > 0) {
          rows = creatorRooms.map((r: any) => ({
            ...r,
            room_id: r.id,
            message_count: r.room_messages?.[0]?.count ?? 0
          }))
        }
      } catch (e) {
        console.error('[persistence.getRoomsForUser] supabase creator rooms exception:', e)
      }

      // 2) Fetch codes claimed by this user and include their rooms (if not already present)
      try {
        const { data: codesData, error: codesError } = await supabase
          .from('codes')
          .select('id, code, room_id')
          .eq('user_id', userId)
          .not('room_id', 'is', null)

        if (codesError) {
          console.error('[persistence.getRoomsForUser] supabase codes fetch error:', codesError)
        } else if (codesData && Array.isArray(codesData) && codesData.length > 0) {
          const roomIds = Array.from(new Set(codesData.map((c: any) => c.room_id).filter(Boolean)))
          if (roomIds.length > 0) {
            const { data: rdata, error: rerr } = await supabase
              .from('rooms')
              .select('*, room_messages(count)')
              .in('id', roomIds)
              .eq('status', 1)
            if (rerr) {
              console.error('[persistence.getRoomsForUser] supabase rooms by id error:', rerr)
            } else if (rdata && Array.isArray(rdata)) {
              const existingIds = new Set(rows.map((rr) => rr.room_id))
              for (const c of codesData) {
                const r = rdata.find((x: any) => x.id === c.room_id)
                if (!r) continue
                const rid = (r as any).id
                if (existingIds.has(rid)) continue
                rows.push({
                  ...(r as any),
                  room_id: rid,
                  code_id: c.id,
                  code: c.code,
                  message_count: (r as any).room_messages?.[0]?.count ?? 0
                })
                existingIds.add(rid)
              }
            }
          }
        }
      } catch (e) {
        console.error('[persistence.getRoomsForUser] supabase codes exception:', e)
      }

      console.log('[persistence.getRoomsForUser] supabase rows=', rows.length)
      return aggregate(rows)
    } catch (e) {
      console.error('[persistence.getRoomsForUser] supabase threw:', e)
    }
  }

  // 3) Fallback (file store or in-memory)
  if (isDevFallbackAllowed()) {
    const store = await pickFallbackStore()
    const rows: Array<Record<string, any>> = []
    for (const c of store.codes) {
      if (c.user_id !== userId || !c.room_id) continue
      const room = store.rooms.find((r) => r.id === c.room_id)
      if (!room || room.status === 0) continue
      const messageCount = store.room_messages.filter((m) => m.room_id === c.room_id).length
      rows.push({
        ...room,
        room_id: room.id,
        code_id: c.id,
        code: c.code,
        message_count: messageCount
      })
    }
    return aggregate(rows)
  }

  return []
}

// Sync: push local rows to Supabase (simple idempotent upsert-like behavior)
export async function syncLocalToSupabase() {
  if (!pool) return { synced: 0 }
  const supabaseUrl = process.env.SUPABASE_URL || process.env.NEXT_PUBLIC_SUPABASE_URL
  if (!supabaseUrl || !process.env.SUPABASE_SERVICE_ROLE_KEY) {
    console.error('Missing Supabase service credentials for sync; skipping push')
    return { synced: 0 }
  }
  const supabase = await createServerClient({ service: true })
  let synced = 0

  // Sync rooms
  const roomsRes = await pool.query('SELECT * FROM rooms')
  for (const r of roomsRes.rows) {
    // Upsert by id
    const { data, error } = await supabase.from('rooms').upsert(r).select().single()
    if (!error) synced++
  }

  // Sync codes
  const codesRes = await pool.query('SELECT * FROM codes')
  for (const c of codesRes.rows) {
    const { data, error } = await supabase.from('codes').upsert(c).select().single()
    if (!error) synced++
  }

  // Sync room_messages
  const messagesRes = await pool.query('SELECT * FROM room_messages')
  for (const m of messagesRes.rows) {
    const { data, error } = await supabase.from('room_messages').upsert(m).select().single()
    if (!error) synced++
  }

  // Sync users can be added similarly
  return { synced }
}

export async function closePool() {
  if (pool) await pool.end()
}

// Messages

async function upsertMessagesLocal(rows: Array<Record<string, any>>) {
  if (!pool || rows.length === 0) return
  for (const row of rows) {
    const cols = Object.keys(row)
    const vals = Object.values(row)
    const colList = cols.map((c) => `"${c}"`).join(',')
    const placeholders = cols.map((_, i) => `$${i + 1}`).join(',')
    const updates = cols.map((c) => `"${c}" = EXCLUDED."${c}"`).join(',')
    const query = `INSERT INTO room_messages (${colList}) VALUES(${placeholders}) ON CONFLICT (id) DO UPDATE SET ${updates}`
    await pool.query(query, vals)
  }
}

export async function syncRoomMessagesFromSupabase(roomId: string) {
  if (!pool) return { synced: 0 }
  if (!process.env.SUPABASE_SERVICE_ROLE_KEY) return { synced: 0 }
  const supabase = await createServerClient({ service: true })
  const { rows: latestRows } = await pool.query('SELECT COALESCE(MAX(created_at), \'1970-01-01\') AS last FROM room_messages WHERE room_id = $1', [roomId])
  const last = latestRows[0]?.last
  const { data, error } = await supabase
    .from('room_messages')
    .select('*')
    .eq('room_id', roomId)
    .gt('created_at', last)
    .order('created_at', { ascending: true })
  if (error || !data || data.length === 0) return { synced: 0 }
  await upsertMessagesLocal(data as any[])
  return { synced: data.length }
}

export async function insertMessage(payload: Database['public']['Tables']['room_messages']['Insert']) {
  const normalized: Database['public']['Tables']['room_messages']['Insert'] = {
    ...payload,
    created_at: payload.created_at || new Date().toISOString()
  }

  if (pool) {
    const cols = Object.keys(normalized)
    const vals = Object.values(normalized)
    const idx = vals.map((_, i) => `$${i + 1}`).join(',')
    const query = `INSERT INTO room_messages(${cols.join(',')}) VALUES(${idx}) RETURNING *`
    const res = await pool.query(query, vals)
    await logEvent({ module: 'messages', operation: 'insert', data: res.rows[0] })
    return res.rows[0]
  }

  try {
    const supabase = await getSupabaseWrite()
    const { data, error } = await supabase.from('room_messages').insert(normalized).select().single()
    if (error) {
      console.error('[persistence.insertMessage] supabase error:', error)
      if (isDevFallbackAllowed()) {
        const id = randomUUID()
        const row = { id, ...normalized }
        inMemory.room_messages.push(row)
        await logEvent({ module: 'messages', operation: 'insert_fallback', data: row, error: error.message || null })
        return row
      }
      return null
    }
    const row = data || null
    await logEvent({ module: 'messages', operation: 'insert', data: row })
    return row
  } catch (e) {
    console.error('[persistence.insertMessage] supabase error (exception):', e)
    if (isDevFallbackAllowed()) {
      const id = randomUUID()
      const row = { id, ...normalized }
      inMemory.room_messages.push(row)
      await logEvent({ module: 'messages', operation: 'insert_fallback', data: row, error: (e as Error)?.message || String(e) })
      return row
    }
    return null
  }
}

export async function getMessagesByRoom(roomId: string, limit = 50) {
  // When running with local DB and service role, try to pull fresh messages from Supabase first
  if (pool && process.env.SUPABASE_SERVICE_ROLE_KEY) {
    try {
      await syncRoomMessagesFromSupabase(roomId)
    } catch (e) {
      console.error('[persistence.getMessagesByRoom] sync from supabase failed (ignored):', e)
    }
  }

  if (pool) {
    const res = await pool.query('SELECT * FROM room_messages WHERE room_id = $1 ORDER BY created_at ASC LIMIT $2', [roomId, limit])
    return res.rows
  }
  try {
    const supabase = await getSupabaseWrite()
    const { data, error } = await supabase
      .from('room_messages')
      .select('*')
      .eq('room_id', roomId)
      .order('created_at', { ascending: true })
      .limit(limit)
    if (error) {
      console.error('[persistence.getMessagesByRoom] supabase error:', error)
      if (isDevFallbackAllowed()) {
        return inMemory.room_messages.filter((m) => m.room_id === roomId).slice(0, limit)
      }
      return []
    }
    return data || []
  } catch (e) {
    console.error('[persistence.getMessagesByRoom] supabase error (exception):', e)
    if (isDevFallbackAllowed()) {
      return inMemory.room_messages.filter((m) => m.room_id === roomId).slice(0, limit)
    }
    return []
  }
}