Complete Planner session management (tasks 4.2, 4.6, 4.7) (#9)

apps/planner/app/lib/db.ts

@@ -0,0 +1,23 @@
+import postgres from "postgres";
+
+const connectionString =
+  process.env.DATABASE_URL ?? "postgres://trails:trails@localhost:5432/trails";
+
+export const sql = postgres(connectionString);
+
+export async function initDb() {
+  await sql`
+    CREATE SCHEMA IF NOT EXISTS planner
+  `;
+  await sql`
+    CREATE TABLE IF NOT EXISTS planner.sessions (
+      id TEXT PRIMARY KEY,
+      yjs_state BYTEA,
+      callback_url TEXT,
+      callback_token TEXT,
+      created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+      last_activity TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+      closed BOOLEAN NOT NULL DEFAULT FALSE
+    )
+  `;
+}

apps/planner/app/lib/sessions.ts

@@ -1,6 +1,7 @@
 import { randomUUID } from "node:crypto";
 import * as Y from "yjs";
 import { getOrCreateDoc, deleteDoc } from "./yjs-server";
+import { sql } from "./db";
 
 export interface SessionMetadata {
   id: string;
@@ -8,45 +9,64 @@ export interface SessionMetadata {
   lastActivity: Date;
   callbackUrl?: string;
   callbackToken?: string;
+  closed: boolean;
 }
 
-const sessions = new Map<string, SessionMetadata>();
-
-export function createSession(options?: {
+export async function createSession(options?: {
   callbackUrl?: string;
   callbackToken?: string;
-}): SessionMetadata {
+}): Promise<SessionMetadata> {
   const id = randomUUID();
-  const now = new Date();
-  const session: SessionMetadata = {
-    id,
-    createdAt: now,
-    lastActivity: now,
-    callbackUrl: options?.callbackUrl,
-    callbackToken: options?.callbackToken,
-  };
+  const doc = getOrCreateDoc(id);
 
-  sessions.set(id, session);
-  getOrCreateDoc(id); // Initialize Yjs doc
+  const [row] = await sql`
+    INSERT INTO planner.sessions (id, yjs_state, callback_url, callback_token)
+    VALUES (${id}, ${Buffer.from(Y.encodeStateAsUpdate(doc))}, ${options?.callbackUrl ?? null}, ${options?.callbackToken ?? null})
+    RETURNING id, created_at, last_activity, callback_url, callback_token, closed
+  `;
 
-  return session;
+  return mapRow(row);
 }
 
-export function getSession(id: string): SessionMetadata | undefined {
-  return sessions.get(id);
+export async function getSession(id: string): Promise<SessionMetadata | undefined> {
+  const [row] = await sql`
+    SELECT id, created_at, last_activity, callback_url, callback_token, closed
+    FROM planner.sessions
+    WHERE id = ${id} AND closed = FALSE
+  `;
+  return row ? mapRow(row) : undefined;
 }
 
-export function touchSession(id: string): void {
-  const session = sessions.get(id);
-  if (session) {
-    session.lastActivity = new Date();
-  }
+export async function touchSession(id: string): Promise<void> {
+  await sql`
+    UPDATE planner.sessions SET last_activity = NOW() WHERE id = ${id}
+  `;
 }
 
-export function closeSession(id: string): boolean {
-  const existed = sessions.delete(id);
+export async function saveSessionState(id: string): Promise<void> {
+  const doc = getOrCreateDoc(id);
+  const state = Y.encodeStateAsUpdate(doc);
+  await sql`
+    UPDATE planner.sessions
+    SET yjs_state = ${Buffer.from(state)}, last_activity = NOW()
+    WHERE id = ${id}
+  `;
+}
+
+export async function loadSessionState(id: string): Promise<Uint8Array | null> {
+  const [row] = await sql`
+    SELECT yjs_state FROM planner.sessions WHERE id = ${id}
+  `;
+  return row?.yjs_state ? new Uint8Array(row.yjs_state) : null;
+}
+
+export async function closeSession(id: string): Promise<boolean> {
+  const result = await sql`
+    UPDATE planner.sessions SET closed = TRUE WHERE id = ${id} AND closed = FALSE
+    RETURNING id
+  `;
   deleteDoc(id);
-  return existed;
+  return result.length > 0;
 }
 
 export function initializeSessionWithWaypoints(
@@ -67,6 +87,36 @@ export function initializeSessionWithWaypoints(
   });
 }
 
-export function listSessions(): SessionMetadata[] {
-  return Array.from(sessions.values());
+export async function listSessions(): Promise<SessionMetadata[]> {
+  const rows = await sql`
+    SELECT id, created_at, last_activity, callback_url, callback_token, closed
+    FROM planner.sessions
+    WHERE closed = FALSE
+    ORDER BY last_activity DESC
+  `;
+  return rows.map(mapRow);
+}
+
+export async function expireSessions(maxAgeDays: number = 7): Promise<number> {
+  const result = await sql`
+    DELETE FROM planner.sessions
+    WHERE last_activity < NOW() - INTERVAL '1 day' * ${maxAgeDays}
+    RETURNING id
+  `;
+  for (const row of result) {
+    deleteDoc(row.id);
+  }
+  return result.length;
+}
+
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+function mapRow(row: any): SessionMetadata {
+  return {
+    id: row.id as string,
+    createdAt: row.created_at as Date,
+    lastActivity: row.last_activity as Date,
+    callbackUrl: (row.callback_url as string) ?? undefined,
+    callbackToken: (row.callback_token as string) ?? undefined,
+    closed: row.closed as boolean,
+  };
 }

apps/planner/app/lib/yjs-server.ts

@@ -1,8 +1,10 @@
 import { WebSocketServer, type WebSocket } from "ws";
 import * as Y from "yjs";
 import type { IncomingMessage, Server } from "node:http";
+import { saveSessionState, loadSessionState, touchSession } from "./sessions";
 
 const docs = new Map<string, Y.Doc>();
+const sessionClients = new Map<string, Set<WebSocket>>();
 
 export function getOrCreateDoc(sessionId: string): Y.Doc {
   let doc = docs.get(sessionId);
@@ -13,11 +15,25 @@ export function getOrCreateDoc(sessionId: string): Y.Doc {
   return doc;
 }
 
+export async function getOrLoadDoc(sessionId: string): Promise<Y.Doc> {
+  let doc = docs.get(sessionId);
+  if (!doc) {
+    doc = new Y.Doc();
+    const savedState = await loadSessionState(sessionId);
+    if (savedState) {
+      Y.applyUpdate(doc, savedState);
+    }
+    docs.set(sessionId, doc);
+  }
+  return doc;
+}
+
 export function deleteDoc(sessionId: string): boolean {
   const doc = docs.get(sessionId);
   if (doc) {
     doc.destroy();
     docs.delete(sessionId);
+    sessionClients.delete(sessionId);
     return true;
   }
   return false;
@@ -27,6 +43,28 @@ export function getDocCount(): number {
   return docs.size;
 }
 
+export function getClientCount(sessionId: string): number {
+  return sessionClients.get(sessionId)?.size ?? 0;
+}
+
+// Debounced persistence — save at most every 5 seconds per session
+const saveTimers = new Map<string, ReturnType<typeof setTimeout>>();
+
+function debouncedSave(sessionId: string): void {
+  if (saveTimers.has(sessionId)) return;
+  saveTimers.set(
+    sessionId,
+    setTimeout(async () => {
+      saveTimers.delete(sessionId);
+      try {
+        await saveSessionState(sessionId);
+      } catch (_e) {
+        // Log but don't crash on save failure
+      }
+    }, 5000),
+  );
+}
+
 export function setupYjsWebSocket(server: Server): WebSocketServer {
   const wss = new WebSocketServer({ noServer: true });
 
@@ -44,13 +82,21 @@ export function setupYjsWebSocket(server: Server): WebSocketServer {
     });
   });
 
-  wss.on("connection", (ws: WebSocket, _request: IncomingMessage, sessionId: string) => {
-    const doc = getOrCreateDoc(sessionId);
+  wss.on("connection", async (ws: WebSocket, _request: IncomingMessage, sessionId: string) => {
+    const doc = await getOrLoadDoc(sessionId);
+
+    // Track clients per session
+    if (!sessionClients.has(sessionId)) {
+      sessionClients.set(sessionId, new Set());
+    }
+    sessionClients.get(sessionId)!.add(ws);
 
     // Send current state to new client
     const state = Y.encodeStateAsUpdate(doc);
     ws.send(state);
 
+    await touchSession(sessionId);
+
     // Listen for updates from this client
     ws.on("message", (data: Buffer) => {
       try {
@@ -58,19 +104,31 @@ export function setupYjsWebSocket(server: Server): WebSocketServer {
         Y.applyUpdate(doc, update);
 
         // Broadcast to all other clients in this session
-        for (const client of wss.clients) {
-          if (client !== ws && client.readyState === ws.OPEN) {
-            client.send(data);
+        const clients = sessionClients.get(sessionId);
+        if (clients) {
+          for (const client of clients) {
+            if (client !== ws && client.readyState === ws.OPEN) {
+              client.send(data);
+            }
           }
         }
+
+        // Persist state (debounced)
+        debouncedSave(sessionId);
       } catch (_e) {
         // Ignore malformed updates
       }
     });
 
     ws.on("close", () => {
-      // Clean up empty sessions
-      // (In production, this would check participant count)
+      const clients = sessionClients.get(sessionId);
+      if (clients) {
+        clients.delete(ws);
+        if (clients.size === 0) {
+          // Last client left — save immediately
+          saveSessionState(sessionId).catch(() => {});
+        }
+      }
     });
   });
 

apps/planner/app/routes/api.sessions.ts

@@ -15,7 +15,7 @@ export async function action({ request }: Route.ActionArgs) {
     gpx?: string;
   };
 
-  const session = createSession({ callbackUrl, callbackToken });
+  const session = await createSession({ callbackUrl, callbackToken });
 
   if (gpx) {
     try {
@@ -36,5 +36,6 @@ export async function action({ request }: Route.ActionArgs) {
 }
 
 export async function loader(_args: Route.LoaderArgs) {
-  return data({ sessions: listSessions() });
+  const sessions = await listSessions();
+  return data({ sessions });
 }

apps/planner/app/routes/session.$id.tsx

@@ -8,7 +8,7 @@ export function meta(_args: Route.MetaArgs) {
 }
 
 export async function loader({ params }: Route.LoaderArgs) {
-  const session = getSession(params.id);
+  const session = await getSession(params.id);
   if (!session) {
     throw data({ error: "Session not found" }, { status: 404 });
   }

apps/planner/package.json

@@ -19,6 +19,7 @@
     "@trails-cool/types": "workspace:*",
     "@trails-cool/ui": "workspace:*",
     "isbot": "^5.1.0",
+    "postgres": "^3.4.8",
     "react": "catalog:",
     "react-dom": "catalog:",
     "react-router": "catalog:",