implement better-auth auth with postgres and route protection

lib/server/auth-session.ts

@@ -0,0 +1,121 @@
+import { headers } from 'next/headers';
+import { ensureAuthSchema } from '@/lib/auth';
+import { asErrorMessage, jsonError } from '@/lib/server/http';
+
+type RecordValue = Record<string, unknown>;
+
+export type AuthenticatedUser = {
+  id: string;
+  email: string;
+  name: string | null;
+  image: string | null;
+  role?: string | string[];
+};
+
+export type AuthenticatedSession = {
+  user: AuthenticatedUser;
+  session: RecordValue | null;
+  raw: RecordValue;
+};
+
+const UNAUTHORIZED_SESSION: AuthenticatedSession = {
+  user: {
+    id: '',
+    email: '',
+    name: null,
+    image: null
+  },
+  session: null,
+  raw: {}
+};
+
+function asRecord(value: unknown): RecordValue | null {
+  if (!value || typeof value !== 'object' || Array.isArray(value)) {
+    return null;
+  }
+
+  return value as RecordValue;
+}
+
+function asString(value: unknown) {
+  return typeof value === 'string' && value.trim().length > 0 ? value : null;
+}
+
+function asNullableString(value: unknown) {
+  return typeof value === 'string' ? value : null;
+}
+
+function normalizeRole(value: unknown) {
+  if (typeof value === 'string') {
+    return value;
+  }
+
+  if (Array.isArray(value)) {
+    const roles = value.filter((entry): entry is string => typeof entry === 'string');
+    return roles.length > 0 ? roles : undefined;
+  }
+
+  return undefined;
+}
+
+function normalizeSession(rawSession: unknown): AuthenticatedSession | null {
+  const root = asRecord(rawSession);
+  if (!root) {
+    return null;
+  }
+
+  const rootSession = asRecord(root.session);
+  const userRecord = asRecord(root.user) ?? asRecord(rootSession?.user);
+  if (!userRecord) {
+    return null;
+  }
+
+  const id = asString(userRecord.id);
+  const email = asString(userRecord.email);
+  if (!id || !email) {
+    return null;
+  }
+
+  return {
+    user: {
+      id,
+      email,
+      name: asNullableString(userRecord.name),
+      image: asNullableString(userRecord.image),
+      role: normalizeRole(userRecord.role)
+    },
+    session: rootSession,
+    raw: root
+  };
+}
+
+export async function getAuthenticatedSession() {
+  const auth = await ensureAuthSchema();
+  const session = await auth.api.getSession({
+    headers: await headers()
+  });
+
+  return normalizeSession(session);
+}
+
+export async function requireAuthenticatedSession() {
+  try {
+    const session = await getAuthenticatedSession();
+    if (!session) {
+      return {
+        session: UNAUTHORIZED_SESSION,
+        response: jsonError('Unauthorized', 401)
+      };
+    }
+
+    return {
+      session,
+      response: null
+    };
+  } catch (error) {
+    return {
+      session: UNAUTHORIZED_SESSION,
+      response: jsonError(asErrorMessage(error, 'Authentication subsystem is unavailable.'), 500)
+    };
+  }
+}

lib/server/store.ts

@@ -21,13 +21,7 @@ const STORE_PATH = path.join(DATA_DIR, 'store.json');
 
 let writeQueue = Promise.resolve();
 
-function nowIso() {
-  return new Date().toISOString();
-}
-
 function createDefaultStore(): DataStore {
-  const now = nowIso();
-
   return {
     counters: {
       watchlist: 0,
@@ -39,19 +33,7 @@ function createDefaultStore(): DataStore {
     holdings: [],
     filings: [],
     tasks: [],
-    insights: [
-      {
-        id: 1,
-        user_id: 1,
-        provider: 'local-bootstrap',
-        model: 'zeroclaw',
-        content: [
-          'System initialized in local-first mode.',
-          'Add holdings and sync filings to produce a live AI brief via OpenClaw.'
-        ].join('\n'),
-        created_at: now
-      }
-    ]
+    insights: []
   };
 }
 

lib/server/tasks.ts

@@ -7,6 +7,7 @@ import { fetchFilingMetrics, fetchRecentFilings } from '@/lib/server/sec';
 import { getStoreSnapshot, withStore } from '@/lib/server/store';
 
 type EnqueueTaskInput = {
+  userId: string;
   taskType: TaskType;
   payload?: Record<string, unknown>;
   priority?: number;
@@ -137,9 +138,15 @@ async function processSyncFilings(task: Task) {
   };
 }
 
-async function processRefreshPrices() {
+async function processRefreshPrices(task: Task) {
+  const userId = task.user_id;
+  if (!userId) {
+    throw new Error('Task is missing user scope');
+  }
+
   const snapshot = await getStoreSnapshot();
-  const tickers = [...new Set(snapshot.holdings.map((holding) => holding.ticker))];
+  const userHoldings = snapshot.holdings.filter((holding) => holding.user_id === userId);
+  const tickers = [...new Set(userHoldings.map((holding) => holding.ticker))];
   const quotes = new Map<string, number>();
 
   for (const ticker of tickers) {
@@ -152,6 +159,10 @@ async function processRefreshPrices() {
 
   await withStore((store) => {
     store.holdings = store.holdings.map((holding) => {
+      if (holding.user_id !== userId) {
+        return holding;
+      }
+
       const quote = quotes.get(holding.ticker);
       if (quote === undefined) {
         return holding;
@@ -236,14 +247,20 @@ function holdingDigest(holdings: Holding[]) {
   }));
 }
 
-async function processPortfolioInsights() {
+async function processPortfolioInsights(task: Task) {
+  const userId = task.user_id;
+  if (!userId) {
+    throw new Error('Task is missing user scope');
+  }
+
   const snapshot = await getStoreSnapshot();
-  const summary = buildPortfolioSummary(snapshot.holdings);
+  const userHoldings = snapshot.holdings.filter((holding) => holding.user_id === userId);
+  const summary = buildPortfolioSummary(userHoldings);
 
   const prompt = [
     'Generate portfolio intelligence with actionable recommendations.',
     `Portfolio summary: ${JSON.stringify(summary)}`,
-    `Holdings: ${JSON.stringify(holdingDigest(snapshot.holdings))}`,
+    `Holdings: ${JSON.stringify(holdingDigest(userHoldings))}`,
     'Respond with: 1) health score (0-100), 2) top 3 risks, 3) top 3 opportunities, 4) next actions in 7 days.'
   ].join('\n');
 
@@ -255,7 +272,7 @@ async function processPortfolioInsights() {
 
     const insight: PortfolioInsight = {
       id: store.counters.insights,
-      user_id: 1,
+      user_id: userId,
       provider: analysis.provider,
       model: analysis.model,
       content: analysis.text,
@@ -277,11 +294,11 @@ async function runTaskProcessor(task: Task) {
     case 'sync_filings':
       return await processSyncFilings(task);
     case 'refresh_prices':
-      return await processRefreshPrices();
+      return await processRefreshPrices(task);
     case 'analyze_filing':
       return await processAnalyzeFiling(task);
     case 'portfolio_insights':
-      return await processPortfolioInsights();
+      return await processPortfolioInsights(task);
     default:
       throw new Error(`Unsupported task type: ${task.task_type}`);
   }
@@ -356,6 +373,7 @@ export async function enqueueTask(input: EnqueueTaskInput) {
 
   const task: Task = {
     id: randomUUID(),
+    user_id: input.userId,
     task_type: input.taskType,
     status: 'queued',
     priority: input.priority ?? 50,
@@ -384,18 +402,19 @@ export async function enqueueTask(input: EnqueueTaskInput) {
   return task;
 }
 
-export async function getTaskById(taskId: string) {
+export async function getTaskById(taskId: string, userId: string) {
   const snapshot = await getStoreSnapshot();
-  return snapshot.tasks.find((task) => task.id === taskId) ?? null;
+  return snapshot.tasks.find((task) => task.id === taskId && task.user_id === userId) ?? null;
 }
 
-export async function listRecentTasks(limit = 20, statuses?: TaskStatus[]) {
+export async function listRecentTasks(userId: string, limit = 20, statuses?: TaskStatus[]) {
   const safeLimit = Math.min(Math.max(Math.trunc(limit), 1), 200);
   const snapshot = await getStoreSnapshot();
+  const scoped = snapshot.tasks.filter((task) => task.user_id === userId);
 
   const filtered = statuses && statuses.length > 0
-    ? snapshot.tasks.filter((task) => statuses.includes(task.status))
-    : snapshot.tasks;
+    ? scoped.filter((task) => statuses.includes(task.status))
+    : scoped;
 
   return filtered
     .slice()