diff --git a/__tests__/api/notifications.test.ts b/__tests__/api/notifications.test.ts
new file mode 100644
index 0000000..a9c05ba
--- /dev/null
+++ b/__tests__/api/notifications.test.ts
@@ -0,0 +1,576 @@
+import { describe, it, expect, vi, beforeEach } from 'vitest'
+import { NextRequest } from 'next/server'
+
+vi.mock('@/lib/db', () => ({
+  sql: vi.fn(),
+}))
+
+vi.mock('@/lib/auth/middleware', async () => {
+  const actual = await vi.importActual<
+    typeof import('@/lib/auth/middleware')
+  >('@/lib/auth/middleware')
+  // Both auth wrappers inject the same fake AuthContext so the handler
+  // body runs as if a valid JWT was presented. Without mocking
+  // withAuthCtx too, PATCH routes (which use withAuthCtx to receive a
+  // Next.js route context) would fall through to the real wrapper and
+  // return 401.
+  const injectAuth = (handler: (...args: unknown[]) => unknown) =>
+    async (request: unknown, ...rest: unknown[]) =>
+    handler(
+      request,
+      { walletAddress: '0xTEST_WALLET', tokenJti: 'test-jti' },
+      ...rest,
+    )
+  return {
+    ...actual,
+    withAuth: injectAuth,
+    withAuthCtx: injectAuth,
+    resolveUserIdByWallet: vi.fn(async (walletAddress: string) => {
+      if (walletAddress === '0xUNRESOLVED') return null
+      return 42
+    }),
+  }
+})
+
+import { sql } from '@/lib/db'
+import { GET as listNotifications } from '@/app/api/notifications/route'
+import {
+  PATCH as markRead,
+} from '@/app/api/notifications/[id]/read/route'
+import {
+  POST as markAllRead,
+} from '@/app/api/notifications/read-all/route'
+import {
+  GET as streamNotifications,
+} from '@/app/api/notifications/stream/route'
+import {
+  NotificationError,
+  NotificationHub,
+  createNotification,
+  listNotificationsForUser,
+  mapNotificationRow,
+  markAllNotificationsRead,
+  markNotificationRead,
+  parseNotificationQuery,
+  notifyContractCreated,
+  notifyMilestoneApproved,
+  notifyMilestoneSubmitted,
+  notifyDisputeCreated,
+  notifyEscrowReleased,
+  NOTIFICATION_EVENT_TYPES,
+  NOTIFICATION_MAX_LIMIT,
+  type Notification,
+} from '@/lib/notifications'
+
+type SqlMock = ReturnType<typeof vi.fn>
+
+interface NotificationRowOverrides {
+  id?: number
+  user_id?: number
+  title?: string
+  message?: string
+  type?: string
+  event_type?: string
+  payload?: Record<string, unknown>
+  is_read?: boolean
+  channel?: string
+  created_at?: Date | string
+  delivered_at?: Date | string | null
+  total_count?: string | number
+}
+
+function buildNotificationRow(
+  overrides: NotificationRowOverrides = {},
+): Array<Record<string, unknown>> {
+  return [
+    {
+      id: 1,
+      user_id: 42,
+      title: 'Milestone Approved',
+      message: 'Milestone "Build login" was approved.',
+      type: 'success',
+      event_type: 'milestone_approved',
+      payload: { milestoneId: 7 },
+      is_read: false,
+      channel: 'in_app',
+      created_at: new Date('2026-01-01T00:00:00Z'),
+      delivered_at: null,
+      total_count: '3',
+      ...overrides,
+    },
+  ]
+}
+
+function queueSqlFail(error: unknown): void {
+  ;(sql as unknown as SqlMock).mockRejectedValueOnce(error)
+}
+
+function queueSql(responses: unknown[]): void {
+  const mock = sql as unknown as SqlMock
+  for (const response of responses) {
+    mock.mockResolvedValueOnce(response)
+  }
+}
+
+function makeRequest(url: string): NextRequest {
+  return new NextRequest(new Request(url))
+}
+
+beforeEach(() => {
+  vi.clearAllMocks()
+  NotificationHub.resetInstance()
+})
+
+describe('parseNotificationQuery', () => {
+  it('returns default values for empty params', () => {
+    expect(parseNotificationQuery(new URLSearchParams())).toEqual({
+      page: 1,
+      limit: 20,
+      type: null,
+      unreadOnly: false,
+    })
+  })
+
+  it('clamps limit to the maximum', () => {
+    const params = parseNotificationQuery(
+      new URLSearchParams(`limit=${NOTIFICATION_MAX_LIMIT * 5}`),
+    )
+    expect(params.limit).toBe(NOTIFICATION_MAX_LIMIT)
+  })
+
+  it('accepts a known event_type', () => {
+    const params = parseNotificationQuery(
+      new URLSearchParams('type=milestone_submitted'),
+    )
+    expect(params.type).toBe('milestone_submitted')
+  })
+
+  it('rejects an unknown event_type', () => {
+    expect(() =>
+      parseNotificationQuery(new URLSearchParams('type=banana')),
+    ).toThrowError(NotificationError)
+  })
+
+  it('accepts truthy unreadOnly variants', () => {
+    expect(
+      parseNotificationQuery(new URLSearchParams('unreadOnly=true')).unreadOnly,
+    ).toBe(true)
+    expect(
+      parseNotificationQuery(new URLSearchParams('unreadOnly=1')).unreadOnly,
+    ).toBe(true)
+  })
+
+  it('rejects invalid page/limit', () => {
+    expect(() => parseNotificationQuery(new URLSearchParams('page=0')))
+      .toThrowError(NotificationError)
+    expect(() => parseNotificationQuery(new URLSearchParams('limit=0')))
+      .toThrowError(NotificationError)
+  })
+
+  it('rejects non-integer page/limit instead of silently defaulting', () => {
+    const pageErr = (() => {
+      try {
+        parseNotificationQuery(new URLSearchParams('page=banana'))
+        return null
+      } catch (e) {
+        return e
+      }
+    })()
+    expect(pageErr).toBeInstanceOf(NotificationError)
+    expect((pageErr as NotificationError).code).toBe('INVALID_PAGE')
+
+    const limitErr = (() => {
+      try {
+        parseNotificationQuery(new URLSearchParams('limit=banana'))
+        return null
+      } catch (e) {
+        return e
+      }
+    })()
+    expect(limitErr).toBeInstanceOf(NotificationError)
+    expect((limitErr as NotificationError).code).toBe('INVALID_LIMIT')
+  })
+
+  it('rejects empty page/limit strings', () => {
+    expect(() => parseNotificationQuery(new URLSearchParams('page=')))
+      .toThrowError(NotificationError)
+    expect(() => parseNotificationQuery(new URLSearchParams('limit=')))
+      .toThrowError(NotificationError)
+  })
+})
+
+describe('mapNotificationRow', () => {
+  it('converts a row into the API shape', () => {
+    const notification = mapNotificationRow({
+      id: 9,
+      user_id: 42,
+      title: 'T',
+      message: 'M',
+      type: 'success',
+      event_type: 'escrow_released',
+      payload: { foo: 'bar' },
+      is_read: false,
+      channel: 'in_app',
+      created_at: new Date('2026-02-02T00:00:00Z'),
+      delivered_at: null,
+    })
+
+    expect(notification).toMatchObject({
+      id: 9,
+      userId: 42,
+      title: 'T',
+      message: 'M',
+      type: 'success',
+      eventType: 'escrow_released',
+      payload: { foo: 'bar' },
+      isRead: false,
+      channel: 'in_app',
+    })
+    expect(notification.createdAt).toMatch(/2026-02-02T/)
+    expect(notification.deliveredAt).toBeNull()
+  })
+})
+
+describe('createNotification', () => {
+  it('persists and publishes to the hub', async () => {
+    queueSql([buildNotificationRow({ id: 101 })])
+
+    const hub = NotificationHub.getInstance()
+    const received: Notification[] = []
+    hub.subscribe(42, (n) => received.push(n))
+
+    const created = await createNotification({
+      userId: 42,
+      title: 'A',
+      message: 'B',
+      type: 'success',
+      eventType: 'milestone_approved',
+      payload: { milestoneId: 1 },
+    })
+
+    expect(created.id).toBe(101)
+    expect(received).toHaveLength(1)
+    expect(received[0].id).toBe(101)
+  })
+
+  it('still returns the persisted notification even if hub publish throws', async () => {
+    queueSql([buildNotificationRow({ id: 102 })])
+
+    const hub = NotificationHub.getInstance()
+    hub.subscribe(42, () => {
+      throw new Error('broken subscriber')
+    })
+
+    const created = await createNotification({
+      userId: 42,
+      title: 'A',
+      message: 'B',
+      type: 'success',
+      eventType: 'milestone_approved',
+      payload: {},
+    })
+
+    expect(created.id).toBe(102)
+  })
+
+  it('throws NotificationError when the INSERT returns no rows', async () => {
+    queueSql([[]])
+    await expect(
+      createNotification({
+        userId: 1,
+        title: 'A',
+        message: 'B',
+        type: 'info',
+        eventType: 'contract_created',
+      }),
+    ).rejects.toBeInstanceOf(NotificationError)
+  })
+})
+
+describe('listNotificationsForUser', () => {
+  it('returns empty result for an out-of-range page', async () => {
+    queueSql([[]]) // main list empty (no COUNT fallback needed since 0 rows handled inline)
+    const result = await listNotificationsForUser(42, {
+      page: 99,
+      limit: 10,
+      type: null,
+      unreadOnly: false,
+    })
+    expect(result.totalItems).toBe(0)
+    expect(result.notifications).toEqual([])
+  })
+
+  it('returns paginated rows with total count via COUNT(*) OVER()', async () => {
+    queueSql([buildNotificationRow({ id: 1, total_count: '5' })])
+    const result = await listNotificationsForUser(42, {
+      page: 1,
+      limit: 1,
+      type: null,
+      unreadOnly: false,
+    })
+    expect(result.totalItems).toBe(5)
+    expect(result.notifications).toHaveLength(1)
+    expect(result.notifications[0].id).toBe(1)
+  })
+
+  it('rejects an invalid user id', async () => {
+    await expect(
+      listNotificationsForUser(0, {
+        page: 1,
+        limit: 10,
+        type: null,
+        unreadOnly: false,
+      }),
+    ).rejects.toBeInstanceOf(NotificationError)
+  })
+
+  it('rejects page/limit pairs whose offset exceeds the DoS cap', async () => {
+    // With page=502, limit=100: offset = (502 - 1) * 100 = 50_100 > 50_000 cap,
+    // so PAGE_TOO_LARGE is thrown before any SQL is issued.
+    await expect(
+      listNotificationsForUser(42, {
+        page: 502,
+        limit: 100,
+        type: null,
+        unreadOnly: false,
+      }),
+    ).rejects.toMatchObject({ code: 'PAGE_TOO_LARGE' })
+  })
+
+  it('accepts the post-cap boundary (page=501, limit=100, offset=50_000)', async () => {
+    // offset == cap -> allowed. The exact cap is inclusive (offset <= cap).
+    queueSql([buildNotificationRow({ id: 1, total_count: '1' })])
+    const result = await listNotificationsForUser(42, {
+      page: 501,
+      limit: 100,
+      type: null,
+      unreadOnly: false,
+    })
+    expect(result.totalItems).toBe(1)
+    expect(result.notifications).toHaveLength(1)
+  })
+})
+
+describe('markNotificationRead', () => {
+  it('returns null when the notification does not belong to the caller', async () => {
+    queueSql([[]])
+    const result = await markNotificationRead(99, 42)
+    expect(result.notification).toBeNull()
+  })
+
+  it('returns the mapped notification when marked', async () => {
+    queueSql([buildNotificationRow({ id: 1, is_read: true })])
+    const result = await markNotificationRead(1, 42)
+    expect(result.notification?.isRead).toBe(true)
+  })
+})
+
+describe('markAllNotificationsRead', () => {
+  it('returns the updated row count', async () => {
+    // markAllNotificationsRead now issues exactly ONE sql call (CTE).
+    queueSql([[{ updated_count: 7 }]])
+    const result = await markAllNotificationsRead(42)
+    expect(result.updatedCount).toBe(7)
+  })
+})
+
+describe('NotificationHub', () => {
+  it('subscriber count drops to zero after unsubscribe', () => {
+    const hub = NotificationHub.getInstance()
+    expect(hub.subscriberCount(42)).toBe(0)
+    const unsub = hub.subscribe(42, () => undefined)
+    expect(hub.subscriberCount(42)).toBe(1)
+    unsub()
+    expect(hub.subscriberCount(42)).toBe(0)
+  })
+
+  it('isolated fan-out: only the targeted user receives the notification', () => {
+    const hub = NotificationHub.getInstance()
+    const seen = { a: 0, b: 0 }
+    hub.subscribe(1, () => {
+      seen.a += 1
+    })
+    hub.subscribe(2, () => {
+      seen.b += 1
+    })
+
+    hub.publish({
+      id: 1,
+      userId: 1,
+      title: 'x',
+      message: 'x',
+      type: 'info',
+      eventType: 'contract_created',
+      payload: {},
+      isRead: false,
+      channel: 'in_app',
+      createdAt: '2026-01-01T00:00:00Z',
+      deliveredAt: null,
+    })
+
+    expect(seen).toEqual({ a: 1, b: 0 })
+  })
+})
+
+describe('GET /api/notifications', () => {
+  it('returns the list with meta', async () => {
+    // resolveUserIdByWallet is mocked, so no sql for the user lookup;
+    // the route issues 2 sql calls (listNotifications + unreadCount).
+    queueSql([
+      buildNotificationRow({ id: 1, total_count: '1' }), // list (with total_count)
+      [{ unread: 1 }], // unread count
+    ])
+
+    const response = await listNotifications(
+      makeRequest('http://localhost/api/notifications?limit=5'),
+    )
+    expect(response.status).toBe(200)
+    const body = await response.json()
+    expect(body.data).toHaveLength(1)
+    expect(body.meta).toMatchObject({
+      totalCount: 1,
+      limit: 5,
+      page: 1,
+      unreadCount: 1,
+    })
+  })
+
+  it('400 with structured code on invalid type', async () => {
+    const response = await listNotifications(
+      makeRequest('http://localhost/api/notifications?type=banana'),
+    )
+    expect(response.status).toBe(400)
+    const body = await response.json()
+    expect(body.code).toBe('INVALID_TYPE')
+  })
+
+  it('503 on DB failure', async () => {
+    queueSqlFail(new Error('connection reset'))
+    const response = await listNotifications(
+      makeRequest('http://localhost/api/notifications'),
+    )
+    expect(response.status).toBe(503)
+    const body = await response.json()
+    expect(body.code).toBe('NOTIFICATIONS_LIST_FAILED')
+  })
+})
+
+describe('PATCH /api/notifications/[id]/read', () => {
+  it('marks the notification read for the caller', async () => {
+    // resolveUserIdByWallet is mocked; the route issue 1 sql call.
+    queueSql([buildNotificationRow({ id: 17, is_read: true })])
+    const response = await markRead(makeRequest('http://localhost/x'), {
+      params: Promise.resolve({ id: '17' }),
+    })
+    expect(response.status).toBe(200)
+    const body = await response.json()
+    expect(body.notification.isRead).toBe(true)
+  })
+
+  it('returns 404 when no row was updated', async () => {
+    queueSql([[]])
+    const response = await markRead(makeRequest('http://localhost/x'), {
+      params: Promise.resolve({ id: '17' }),
+    })
+    expect(response.status).toBe(404)
+    expect((await response.json()).code).toBe('NOT_FOUND')
+  })
+
+  it('returns 400 for a non-numeric id', async () => {
+    const response = await markRead(makeRequest('http://localhost/x'), {
+      params: Promise.resolve({ id: 'banana' }),
+    })
+    expect(response.status).toBe(400)
+    expect((await response.json()).code).toBe('INVALID_ID')
+  })
+
+  it('returns 404 when the wallet does not map to a user', async () => {
+    // Switch the mocked resolveUserIdByWallet to return null for this test.
+    const { resolveUserIdByWallet } = await import('@/lib/auth/middleware')
+    ;(resolveUserIdByWallet as ReturnType<typeof vi.fn>).mockResolvedValueOnce(
+      null,
+    )
+
+    const response = await markRead(makeRequest('http://localhost/x'), {
+      params: Promise.resolve({ id: '17' }),
+    })
+    expect(response.status).toBe(404)
+    expect((await response.json()).code).toBe('USER_NOT_FOUND')
+  })
+})
+
+describe('POST /api/notifications/read-all', () => {
+  it('returns the updated count', async () => {
+    // resolveUserIdByWallet is mocked, so no sql call for the user
+    // lookup; the route then issues a single CTE sql call.
+    queueSql([[{ updated_count: 4 }]])
+    const response = await markAllRead(makeRequest('http://localhost/x'))
+    expect(response.status).toBe(200)
+    expect((await response.json()).updatedCount).toBe(4)
+  })
+})
+
+describe('GET /api/notifications/stream', () => {
+  it('returns a text/event-stream response for the authenticated user', async () => {
+    // resolveUserIdByWallet is mocked to return 42; no sql call needed.
+    const response = await streamNotifications(
+      makeRequest('http://localhost/api/notifications/stream'),
+    )
+
+    expect(response.status).toBe(200)
+    expect(response.headers.get('content-type')).toMatch(/text\/event-stream/)
+    expect(NotificationHub.getInstance().subscriberCount(42)).toBe(1)
+  })
+
+  it('drops subscribers on stream cancel', async () => {
+    const response = await streamNotifications(
+      makeRequest('http://localhost/api/notifications/stream'),
+    )
+    expect(response.status).toBe(200)
+    expect(NotificationHub.getInstance().subscriberCount(42)).toBe(1)
+
+    await response.body?.cancel()
+    // give the cancel() callback a tick to run
+    await Promise.resolve()
+    expect(NotificationHub.getInstance().subscriberCount(42)).toBe(0)
+  })
+})
+
+describe('event-creation helpers', () => {
+  it('notifyContractCreated emits one notification per recipient', async () => {
+    queueSql([
+      buildNotificationRow({ id: 1, user_id: 11, event_type: 'contract_created' }),
+      buildNotificationRow({ id: 2, user_id: 22, event_type: 'contract_created' }),
+    ])
+    await notifyContractCreated(11, 22, 99, 'Logo redesign')
+    expect(NOTIFICATION_EVENT_TYPES).toContain('contract_created')
+  })
+
+  it('notifyMilestoneSubmitted creates exactly one notification', async () => {
+    queueSql([buildNotificationRow({ event_type: 'milestone_submitted' })])
+    await notifyMilestoneSubmitted(11, 5, 'Build login')
+  })
+
+  it('notifyMilestoneApproved creates exactly one notification', async () => {
+    queueSql([buildNotificationRow({ event_type: 'milestone_approved' })])
+    await notifyMilestoneApproved(11, 5, 'Build login')
+  })
+
+  it('notifyEscrowReleased skips the freelancer when null', async () => {
+    queueSql([buildNotificationRow({ event_type: 'escrow_released' })])
+    await notifyEscrowReleased(11, null, 7, '5.00', 'XLM')
+  })
+
+  it('notifyEscrowReleased includes the freelancer when provided', async () => {
+    queueSql([
+      buildNotificationRow({ id: 1, user_id: 11, event_type: 'escrow_released' }),
+      buildNotificationRow({ id: 2, user_id: 22, event_type: 'escrow_released' }),
+    ])
+    await notifyEscrowReleased(11, 22, 7, '5.00', 'XLM')
+  })
+
+  it('notifyDisputeCreated creates exactly one notification', async () => {
+    queueSql([buildNotificationRow({ event_type: 'dispute_created' })])
+    await notifyDisputeCreated(11, 3, 7)
+  })
+})
diff --git a/app/api/notifications/[id]/read/route.ts b/app/api/notifications/[id]/read/route.ts
new file mode 100644
index 0000000..d311360
--- /dev/null
+++ b/app/api/notifications/[id]/read/route.ts
@@ -0,0 +1,85 @@
+import { NextRequest, NextResponse } from 'next/server'
+
+import {
+  NotificationError,
+  markNotificationRead,
+} from '@/lib/notifications'
+import { sql } from '@/lib/db'
+import { withAuthCtx, AuthContext, resolveUserIdByWallet as resolveUserId } from '@/lib/auth/middleware'
+
+export const dynamic = 'force-dynamic'
+
+type RouteContext = { params: Promise<{ id: string }> }
+
+/**
+ * PATCH /api/notifications/[id]/read
+ *
+ * Marks a single notification as read for the authenticated user. The
+ * notification must belong to the caller — otherwise we return 404 to avoid
+ * leaking whether the id exists for someone else.
+ *
+ * Status codes:
+ *   200 ok        {notification}
+ *   400 invalid id or NotificationError → 400
+ *   401 auth required
+ *   404 not found / not the caller's notification
+ *   503 db failure
+ */
+export const PATCH = withAuthCtx<RouteContext>(
+  async (request: NextRequest, auth: AuthContext, context: RouteContext) => {
+    void request
+    const { id: rawId } = await context.params
+    const notificationId = Number.parseInt(rawId, 10)
+
+    if (!Number.isFinite(notificationId) || notificationId <= 0) {
+      return NextResponse.json(
+        { error: 'Invalid notification id', code: 'INVALID_ID' },
+        { status: 400 },
+      )
+    }
+
+    try {
+      const userId = await resolveUserId(auth.walletAddress)
+      if (userId === null) {
+        return NextResponse.json(
+          { error: 'User not found', code: 'USER_NOT_FOUND' },
+          { status: 404 },
+        )
+      }
+
+      const result = await markNotificationRead(notificationId, userId)
+      if (result.notification === null) {
+        return NextResponse.json(
+          { error: 'Notification not found', code: 'NOT_FOUND' },
+          { status: 404 },
+        )
+      }
+
+      return NextResponse.json(
+        { notification: result.notification },
+        {
+          headers: {
+            'Cache-Control': 'no-store',
+          },
+        },
+      )
+    } catch (error) {
+      if (error instanceof NotificationError) {
+        return NextResponse.json(
+          { error: error.message, code: error.code },
+          { status: 400 },
+        )
+      }
+      console.error(
+        `Failed to mark notification ${notificationId} read:`,
+        error,
+      )
+      return NextResponse.json(
+        { error: 'Unable to mark notification read', code: 'NOTIFICATION_UPDATE_FAILED' },
+        { status: 503 },
+      )
+    }
+  },
+)
+
+
diff --git a/app/api/notifications/read-all/route.ts b/app/api/notifications/read-all/route.ts
new file mode 100644
index 0000000..646081a
--- /dev/null
+++ b/app/api/notifications/read-all/route.ts
@@ -0,0 +1,53 @@
+import { NextRequest, NextResponse } from 'next/server'
+
+import {
+  NotificationError,
+  markAllNotificationsRead,
+} from '@/lib/notifications'
+import {
+  withAuth,
+  AuthContext,
+  resolveUserIdByWallet as resolveUserId,
+} from '@/lib/auth/middleware'
+
+export const dynamic = 'force-dynamic'
+
+/**
+ * POST /api/notifications/read-all
+ *
+ * Marks every unread notification for the authenticated user as read.
+ * Returns the number of rows updated (returns 0 if there were none).
+ *
+ * Status codes:
+ *   200 ok            {updatedCount}
+ *   400 NotificationError
+ *   401 auth required
+ *   503 db failure
+ */
+export const POST = withAuth(async (request: NextRequest, auth: AuthContext) => {
+  void request
+  try {
+    const userId = await resolveUserId(auth.walletAddress)
+    if (userId === null) {
+      return NextResponse.json(
+        { error: 'User not found', code: 'USER_NOT_FOUND' },
+        { status: 404 },
+      )
+    }
+
+    const result = await markAllNotificationsRead(userId)
+    return NextResponse.json({ updatedCount: result.updatedCount })
+  } catch (error) {
+    if (error instanceof NotificationError) {
+      return NextResponse.json(
+        { error: error.message, code: error.code },
+        { status: 400 },
+      )
+    }
+    console.error('Failed to mark all notifications read:', error)
+    return NextResponse.json(
+      { error: 'Unable to mark all notifications read', code: 'NOTIFICATION_UPDATE_FAILED' },
+      { status: 503 },
+    )
+  }
+})
diff --git a/app/api/notifications/route.ts b/app/api/notifications/route.ts
index 608db13..ab2dee6 100644
--- a/app/api/notifications/route.ts
+++ b/app/api/notifications/route.ts
@@ -1,89 +1,57 @@
 export const dynamic = "force-dynamic";
 
 import { NextRequest, NextResponse } from "next/server";
-import { z } from "zod";
-import { withAuth } from "@/lib/auth/middleware";
-import { enforceRateLimit, buildRateLimitKey } from "@/lib/security/rateLimit";
-import { getUserIdByWallet } from "@/lib/reputation";
+import { withAuth, AuthContext, resolveUserIdByWallet } from "@/lib/auth/middleware";
 import {
-  listNotifications as listNotificationsDb,
-  countNotifications as countNotificationsDb,
-  markAllAsRead as markAllAsReadDb,
+  parseNotificationQuery,
+  listNotificationsForUser,
+  markAllNotificationsRead,
+  getUnreadCount,
+  NotificationError,
 } from "@/lib/notifications";
 
-// ─── Validation schemas ────────────────────────────────────────────────────
-
-const ListNotificationsSchema = z.object({
-  isRead: z.coerce.boolean().optional(),
-  limit: z.coerce.number().int().min(1).max(100).optional(),
-  offset: z.coerce.number().int().min(0).optional(),
-});
-
 // ─── GET /api/notifications ────────────────────────────────────────────────
 
 /**
- * GET /api/notifications?limit=20&offset=0&isRead=false
+ * GET /api/notifications?page=1&limit=20&type=milestone_approved&unreadOnly=true
  *
  * Returns a paginated list of notifications for the authenticated user.
  * Query parameters:
- *   - limit: max 100, default 20
- *   - offset: pagination offset, default 0
- *   - isRead: filter by read status (optional)
+ *   - page: ≥1, default 1
+ *   - limit: 1..100, default 20
+ *   - type: one of NOTIFICATION_EVENT_TYPES, optional
+ *   - unreadOnly: true/1, optional
+ *
+ * Response:
+ *   { data: Notification[], meta: { totalCount, page, limit, totalPages, unreadCount } }
  */
-export const GET = withAuth(async (request: NextRequest, auth) => {
-  const limited = await enforceRateLimit(request, {
-    key: buildRateLimitKey(request, "notifications:list", auth.walletAddress),
-    limit: 60,
-    windowMs: 60_000,
-  });
-  if (limited) return limited;
-
-  const userId = await getUserIdByWallet(auth.walletAddress);
-  if (userId === null) {
-    return NextResponse.json(
-      {
-        error: "Platform user not found for this wallet",
-        code: "USER_NOT_FOUND",
-      },
-      { status: 404 },
-    );
-  }
-
-  const { searchParams } = request.nextUrl;
-  const parsed = ListNotificationsSchema.safeParse({
-    isRead: searchParams.get("isRead") ?? undefined,
-    limit: searchParams.get("limit") ?? undefined,
-    offset: searchParams.get("offset") ?? undefined,
-  });
-
-  if (!parsed.success) {
-    return NextResponse.json(
-      {
-        error: "Validation failed",
-        details: parsed.error.flatten().fieldErrors,
-      },
-      { status: 422 },
-    );
-  }
-
+export const GET = withAuth(async (request: NextRequest, auth: AuthContext) => {
+  void request;
   try {
-    const notifications = await listNotificationsDb({
-      userId,
-      isRead: parsed.data.isRead,
-      limit: parsed.data.limit,
-      offset: parsed.data.offset,
-    });
+    const userId = await resolveUserIdByWallet(auth.walletAddress);
+    if (userId === null) {
+      return NextResponse.json(
+        { error: "Platform user not found for this wallet", code: "USER_NOT_FOUND" },
+        { status: 404 },
+      );
+    }
+
+    const query = parseNotificationQuery(request.nextUrl.searchParams);
 
-    const total = await countNotificationsDb(userId, parsed.data.isRead);
+    const [result, unreadCount] = await Promise.all([
+      listNotificationsForUser(userId, query),
+      getUnreadCount(userId),
+    ]);
 
     return NextResponse.json(
       {
-        notifications,
-        pagination: {
-          total,
-          limit: parsed.data.limit ?? 20,
-          offset: parsed.data.offset ?? 0,
-          hasMore: (parsed.data.offset ?? 0) + notifications.length < total,
+        data: result.notifications,
+        meta: {
+          totalCount: result.totalItems,
+          page: query.page,
+          limit: query.limit,
+          totalPages: Math.max(1, Math.ceil(result.totalItems / query.limit)),
+          unreadCount,
         },
       },
       {
@@ -92,10 +60,16 @@ export const GET = withAuth(async (request: NextRequest, auth) => {
       },
     );
   } catch (err) {
+    if (err instanceof NotificationError) {
+      return NextResponse.json(
+        { error: err.message, code: err.code },
+        { status: 400 },
+      );
+    }
     console.error("[GET /api/notifications]", err);
     return NextResponse.json(
-      { error: "Failed to fetch notifications" },
-      { status: 500 },
+      { error: "Failed to fetch notifications", code: "NOTIFICATIONS_LIST_FAILED" },
+      { status: 503 },
     );
   }
 });
@@ -106,40 +80,35 @@ export const GET = withAuth(async (request: NextRequest, auth) => {
  * PATCH /api/notifications
  *
  * Marks all unread notifications for the authenticated user as read.
+ * Returns { updatedCount }.
  */
-export const PATCH = withAuth(async (request: NextRequest, auth) => {
-  const limited = await enforceRateLimit(request, {
-    key: buildRateLimitKey(request, "notifications:update", auth.walletAddress),
-    limit: 30,
-    windowMs: 60_000,
-  });
-  if (limited) return limited;
-
-  const userId = await getUserIdByWallet(auth.walletAddress);
-  if (userId === null) {
-    return NextResponse.json(
-      {
-        error: "Platform user not found for this wallet",
-        code: "USER_NOT_FOUND",
-      },
-      { status: 404 },
-    );
-  }
-
+export const PATCH = withAuth(async (request: NextRequest, auth: AuthContext) => {
+  void request;
   try {
-    const updatedCount = await markAllAsReadDb(userId);
+    const userId = await resolveUserIdByWallet(auth.walletAddress);
+    if (userId === null) {
+      return NextResponse.json(
+        { error: "Platform user not found for this wallet", code: "USER_NOT_FOUND" },
+        { status: 404 },
+      );
+    }
+
+    const result = await markAllNotificationsRead(userId);
     return NextResponse.json(
-      {
-        message: "All notifications marked as read",
-        updatedCount,
-      },
+      { message: "All notifications marked as read", updatedCount: result.updatedCount },
       { status: 200 },
     );
   } catch (err) {
+    if (err instanceof NotificationError) {
+      return NextResponse.json(
+        { error: err.message, code: err.code },
+        { status: 400 },
+      );
+    }
     console.error("[PATCH /api/notifications]", err);
     return NextResponse.json(
-      { error: "Failed to update notifications" },
-      { status: 500 },
+      { error: "Failed to update notifications", code: "NOTIFICATION_UPDATE_FAILED" },
+      { status: 503 },
     );
   }
 });
diff --git a/app/api/notifications/stream/route.ts b/app/api/notifications/stream/route.ts
new file mode 100644
index 0000000..3741964
--- /dev/null
+++ b/app/api/notifications/stream/route.ts
@@ -0,0 +1,152 @@
+import { NextRequest, NextResponse } from 'next/server'
+import { randomUUID } from 'node:crypto'
+
+import {
+  NotificationError,
+  NotificationHub,
+  type Notification,
+} from '@/lib/notifications'
+import {
+  withAuth,
+  AuthContext,
+  resolveUserIdByWallet as resolveUserId,
+} from '@/lib/auth/middleware'
+
+export const dynamic = 'force-dynamic'
+export const runtime = 'nodejs'
+
+const KEEP_ALIVE_MS = 25_000
+
+/**
+ * GET /api/notifications/stream
+ *
+ * Server-Sent Events stream that pushes notifications for the authenticated
+ * user in real time. The connection also emits a keep-alive comment every
+ * 25 s so reverse proxies don't close idle sockets.
+ *
+ * Event format:
+ *   id: <notification-id>
+ *   event: notification
+ *   data: { "id": <number>, "userId": <number>, "title": "...", ... }
+ *
+ * Query parameters:
+ *   lastEventId  Optional. Reserved for future replay support — when the
+ *                 client reconnects with the last id it saw, we can replay
+ *                 buffered events. The in-process hub currently has no
+ *                 persistent buffer, so we deliberately do NOT parse it
+ *                 yet (no replay is performed until the buffer ships).
+ *
+ * Status codes:
+ *   200 stream open
+ *   401 missing/invalid auth
+ *   500 if the runtime cannot construct a TextEncoder/Stream
+ */
+export const GET = withAuth(async (request: NextRequest, auth: AuthContext) => {
+  try {
+    const userId = await resolveUserId(auth.walletAddress)
+    if (userId === null) {
+      return NextResponse.json(
+        { error: 'User not found', code: 'USER_NOT_FOUND' },
+        { status: 404 },
+      )
+    }
+
+    void request // reserved for future lastEventId replay parameter
+
+    const encoder = new TextEncoder()
+    const hub = NotificationHub.getInstance()
+
+    // Per-stream state shared by the `start` and `cancel` paths. Using an
+    // interface + arrow-function assignment (rather than an object-literal
+    // method) keeps `this` unambiguous and resilient against future
+    // refactors that pass `state.release` around.
+    interface StreamState {
+      unsub: null | (() => void)
+      keepAlive: null | ReturnType<typeof setInterval>
+      closed: boolean
+      release: () => void
+    }
+    const state: StreamState = {
+      unsub: null,
+      keepAlive: null,
+      closed: false,
+      release: () => undefined,
+    }
+    state.release = () => {
+      if (state.closed) return
+      state.closed = true
+      if (state.unsub) state.unsub()
+      if (state.keepAlive) clearInterval(state.keepAlive)
+    }
+
+    const stream = new ReadableStream<Uint8Array>({
+      start(controller) {
+        const send = (eventName: string, data: unknown, id?: number): void => {
+          try {
+            const lines: string[] = []
+            if (id !== undefined) lines.push(`id: ${id}`)
+            lines.push(`event: ${eventName}`)
+            lines.push(`data: ${JSON.stringify(data)}`)
+            lines.push('', '')
+            controller.enqueue(encoder.encode(lines.join('\n')))
+          } catch (error) {
+            console.error('[notifications] SSE enqueue failed:', error)
+          }
+        }
+
+        send('hello', { ts: new Date().toISOString(), tag: randomUUID() })
+
+        state.unsub = hub.subscribe(userId, (notification: Notification) => {
+          send('notification', notification, notification.id)
+        })
+
+        state.keepAlive = setInterval(() => {
+          try {
+            controller.enqueue(
+              encoder.encode(`: keep-alive ${Date.now()}\n\n`),
+            )
+          } catch {
+            // Controller is already closed; the release path will fire.
+          }
+        }, KEEP_ALIVE_MS)
+
+        // Underlying socket closed.
+        request.signal.addEventListener('abort', () => {
+          state.release()
+          try {
+            controller.close()
+          } catch {
+            // already closed
+          }
+        })
+      },
+      cancel() {
+        // Consumer cancelled the body of the response (e.g. EventSource
+        // closed). Same release path: unsubscribe from the hub and clear
+        // the keep-alive interval.
+        state.release()
+      },
+    })
+
+    return new NextResponse(stream, {
+      headers: {
+        'Content-Type': 'text/event-stream; charset=utf-8',
+        'Cache-Control': 'no-cache, no-transform',
+        Connection: 'keep-alive',
+        'X-Accel-Buffering': 'no',
+      },
+    })
+  } catch (error) {
+    if (error instanceof NotificationError) {
+      return NextResponse.json(
+        { error: error.message, code: error.code },
+        { status: 400 },
+      )
+    }
+    console.error('Failed to open notification stream:', error)
+    return NextResponse.json(
+      { error: 'Unable to open stream', code: 'STREAM_FAILED' },
+      { status: 500 },
+    )
+  }
+})
diff --git a/lib/auth/middleware.ts b/lib/auth/middleware.ts
index 3dbcaff..b8e6cfa 100644
--- a/lib/auth/middleware.ts
+++ b/lib/auth/middleware.ts
@@ -1,5 +1,6 @@
 import { NextRequest, NextResponse } from 'next/server'
 import { readAccessToken, verifyAccessToken } from '@/lib/auth/session'
+import { sql } from '@/lib/db'
 
 export interface AuthContext {
   walletAddress: string
@@ -11,6 +12,12 @@ type AuthenticatedHandler = (
   auth: AuthContext
 ) => Promise<NextResponse> | NextResponse
 
+type AuthenticatedHandlerWithCtx<Ctx> = (
+  request: NextRequest,
+  auth: AuthContext,
+  context: Ctx
+) => Promise<NextResponse> | NextResponse
+
 function unauthorizedResponse(): NextResponse {
   return NextResponse.json(
     { error: 'Unauthorized', code: 'AUTH_REQUIRED' },
@@ -36,3 +43,45 @@ export function withAuth(handler: AuthenticatedHandler) {
     })
   }
 }
+
+export function withAuthCtx<Ctx>(
+  handler: AuthenticatedHandlerWithCtx<Ctx>,
+) {
+  return async (
+    request: NextRequest,
+    context: Ctx,
+  ): Promise<NextResponse> => {
+    const token = readAccessToken(request)
+    if (!token) {
+      return unauthorizedResponse()
+    }
+
+    const payload = verifyAccessToken(token)
+    if (!payload) {
+      return unauthorizedResponse()
+    }
+
+    return handler(request, {
+      walletAddress: payload.walletAddress,
+      tokenJti: payload.jti,
+    }, context)
+  }
+}
+
+/**
+ * Resolves the integer user.id for the given wallet_address. Returns null if
+ * the wallet does not map to a row in `users`. Shared between the
+ * notification routes and any other call site that needs the
+ * authenticated user's database id (the JWT only carries the wallet
+ * address).
+ */
+export async function resolveUserIdByWallet(
+  walletAddress: string,
+): Promise<number | null> {
+  const rows = (await sql`
+    SELECT id FROM users WHERE wallet_address = ${walletAddress} LIMIT 1
+  `) as Array<{ id: number }>
+  if (rows.length === 0) return null
+  const raw = rows[0].id
+  return typeof raw === 'number' ? raw : Number(raw) || null
+}
diff --git a/lib/notifications.ts b/lib/notifications.ts
index 432e8b1..b1a12ef 100644
--- a/lib/notifications.ts
+++ b/lib/notifications.ts
@@ -1,8 +1,9 @@
 // lib/notifications.ts
 //
-// Service layer for notification operations: listing, marking as read/unread,
-// and deleting notifications. All DB access goes through this file so route
-// handlers stay thin and logic is independently testable.
+// Service layer for notification operations: listing, marking as read,
+// real-time push via NotificationHub, and event-creation helpers.
+// All DB access goes through this file so route handlers stay thin and
+// logic is independently testable.
 //
 // Column mapping:
 //   DB snake_case  ←→  JS camelCase (done manually — no ORM)
@@ -28,77 +29,440 @@ export type NotificationType =
   | "wallet_activity";
 
 export interface Notification {
-  id: string;
-  userId: string;
-  type: NotificationType;
+  id: number;
+  userId: number;
+  title: string;
+  message: string;
+  type: string;
+  eventType: string;
   payload: Record<string, unknown>;
   isRead: boolean;
+  channel: string;
   createdAt: string;
+  deliveredAt: string | null;
 }
 
-export interface ListNotificationsFilter {
-  userId: string;
-  isRead?: boolean;
-  limit?: number;
-  offset?: number;
+export interface NotificationQuery {
+  page: number;
+  limit: number;
+  type: string | null;
+  unreadOnly: boolean;
+}
+
+export interface CreateNotificationInput {
+  userId: number;
+  title: string;
+  message: string;
+  type: string;
+  eventType: string;
+  payload?: Record<string, unknown>;
+}
+
+// ─── NotificationError ─────────────────────────────────────────────────────
+
+export class NotificationError extends Error {
+  public code: string;
+
+  constructor(code: string, message: string) {
+    super(message);
+    this.name = "NotificationError";
+    this.code = code;
+  }
+}
+
+// ─── Constants ─────────────────────────────────────────────────────────────
+
+export const NOTIFICATION_EVENT_TYPES = [
+  "info",
+  "success",
+  "warning",
+  "contract_created",
+  "milestone_submitted",
+  "milestone_approved",
+  "escrow_released",
+  "escrow_refunded",
+  "dispute_created",
+  "dispute_resolved",
+] as const;
+
+export type NotificationEventType =
+  (typeof NOTIFICATION_EVENT_TYPES)[number];
+
+export const NOTIFICATION_MAX_LIMIT = 100;
+
+/**
+ * Maximum offset allowed for pagination scans. When (page - 1) * limit
+ * exceeds this cap the query is rejected before touching Postgres so a
+ * malicious client cannot force a huge OFFSET scan.
+ *
+ * 100 * 500 = 50 000 rows
+ */
+export const NOTIFICATION_MAX_OFFSET = NOTIFICATION_MAX_LIMIT * 500;
+
+// ─── NotificationHub (singleton, in-process pub/sub) ───────────────────────
+
+type SubscriberCallback = (notification: Notification) => void;
+
+export class NotificationHub {
+  private static instance: NotificationHub | null = null;
+
+  private readonly subscribers = new Map<number, Set<SubscriberCallback>>();
+
+  static getInstance(): NotificationHub {
+    if (!NotificationHub.instance) {
+      NotificationHub.instance = new NotificationHub();
+    }
+    return NotificationHub.instance;
+  }
+
+  static resetInstance(): void {
+    NotificationHub.instance = null;
+  }
+
+  /**
+   * Subscribe to notifications for a given user. Returns an unsubscribe
+   * function. Safe to call unsubscribe multiple times — subsequent calls
+   * are no-ops.
+   */
+  subscribe(
+    userId: number,
+    callback: SubscriberCallback,
+  ): () => void {
+    let set = this.subscribers.get(userId);
+    if (!set) {
+      set = new Set();
+      this.subscribers.set(userId, set);
+    }
+    set.add(callback);
+
+    let unsubscribed = false;
+    return () => {
+      if (unsubscribed) return;
+      unsubscribed = true;
+      const s = this.subscribers.get(userId);
+      if (s) {
+        s.delete(callback);
+        if (s.size === 0) {
+          this.subscribers.delete(userId);
+        }
+      }
+    };
+  }
+
+  /**
+   * Publish a notification to all subscribers for the notification's
+   * userId. Broken subscribers are isolated — a throw inside one
+   * callback does not prevent other callbacks from running.
+   */
+  publish(notification: Notification): void {
+    const set = this.subscribers.get(notification.userId);
+    if (!set) return;
+    for (const cb of set) {
+      try {
+        cb(notification);
+      } catch (err) {
+        console.error(
+          "[NotificationHub] subscriber error for user",
+          notification.userId,
+          ":",
+          err,
+        );
+      }
+    }
+  }
+
+  /**
+   * Returns the number of active subscribers for a given user.
+   */
+  subscriberCount(userId: number): number {
+    return this.subscribers.get(userId)?.size ?? 0;
+  }
+}
+
+// ─── Query parsing ─────────────────────────────────────────────────────────
+
+/**
+ * Parses URL search params for the GET /api/notifications endpoint.
+ * Throws NotificationError with documented codes on invalid input.
+ */
+export function parseNotificationQuery(
+  params: URLSearchParams,
+): NotificationQuery {
+  const parsePage = (raw: string | null): number => {
+    if (raw === null) {
+      return 1;
+    }
+    // Empty string (e.g. `?page=`) must throw, not silently default.
+    const n = Number.parseInt(raw, 10);
+    if (!Number.isFinite(n) || n < 1) {
+      throw new NotificationError(
+        "INVALID_PAGE",
+        "page must be a positive integer",
+      );
+    }
+    return n;
+  };
+
+  const parseLimit = (raw: string | null): number => {
+    if (raw === null) {
+      return 20;
+    }
+    // Empty string (e.g. `?limit=`) must throw, not silently default.
+    const n = Number.parseInt(raw, 10);
+    if (!Number.isFinite(n) || n < 1) {
+      throw new NotificationError(
+        "INVALID_LIMIT",
+        "limit must be a positive integer",
+      );
+    }
+    return Math.min(n, NOTIFICATION_MAX_LIMIT);
+  };
+
+  const page = parsePage(params.get("page"));
+  const limit = parseLimit(params.get("limit"));
+
+  const typeRaw = params.get("type");
+  const type: string | null = typeRaw ?? null;
+  if (
+    type !== null &&
+    !(NOTIFICATION_EVENT_TYPES as readonly string[]).includes(type)
+  ) {
+    throw new NotificationError(
+      "INVALID_TYPE",
+      `Unknown notification type: ${type}`,
+    );
+  }
+
+  const unreadRaw = params.get("unreadOnly");
+  const unreadOnly =
+    unreadRaw === "true" || unreadRaw === "1";
+
+  return { page, limit, type, unreadOnly };
 }
 
 // ─── Row → domain mapper ───────────────────────────────────────────────────
 
-function rowToNotification(row: Record<string, unknown>): Notification {
+export function mapNotificationRow(
+  row: Record<string, unknown>,
+): Notification {
   return {
-    id: row.id as string,
-    userId: row.user_id as string,
-    type: row.type as NotificationType,
+    id: row.id as number,
+    userId: row.user_id as number,
+    title: row.title as string,
+    message: row.message as string,
+    type: row.type as string,
+    eventType: row.event_type as string,
     payload: (row.payload as Record<string, unknown>) ?? {},
     isRead: row.is_read as boolean,
-    createdAt: (row.created_at as Date).toISOString(),
+    channel: (row.channel as string) ?? "in_app",
+    createdAt:
+      row.created_at instanceof Date
+        ? row.created_at.toISOString()
+        : (row.created_at as string),
+    deliveredAt:
+      row.delivered_at instanceof Date
+        ? row.delivered_at.toISOString()
+        : (row.delivered_at as string | null),
   };
 }
 
-// ─── Service functions ─────────────────────────────────────────────────────
+// ─── Persistence ───────────────────────────────────────────────────────────
 
 /**
- * Returns a paginated list of notifications for a user, optionally filtered
- * by read status. Ordered by created_at descending (newest first).
+ * Persists a notification row and best-effort publishes to the hub.
+ * Throws NotificationError if the INSERT returns zero rows.
  */
-export async function listNotifications(
-  filter: ListNotificationsFilter,
-): Promise<Notification[]> {
-  const limit = Math.min(filter.limit ?? 20, 100); // hard cap at 100
-  const offset = filter.offset ?? 0;
+export async function createNotification(
+  input: CreateNotificationInput,
+): Promise<Notification> {
+  const rows = (await sql`
+    INSERT INTO notifications
+      (user_id, title, message, type, event_type, payload)
+    VALUES
+      (${input.userId}, ${input.title}, ${input.message}, ${input.type},
+       ${input.eventType}, ${JSON.stringify(input.payload ?? {})})
+    RETURNING *
+  `) as Record<string, unknown>[];
 
-  // Build WHERE clauses dynamically
+  if (rows.length === 0) {
+    throw new NotificationError(
+      "NOTIFICATION_CREATE_FAILED",
+      "No row returned after INSERT",
+    );
+  }
+
+  const notification = mapNotificationRow(rows[0]);
+
+  // Best-effort hub publish — failure here must not throw.
+  try {
+    NotificationHub.getInstance().publish(notification);
+  } catch (err) {
+    console.error(
+      "[notifications] hub publish failed (id=",
+      notification.id,
+      "):",
+      err,
+    );
+  }
+
+  return notification;
+}
+
+// ─── List ──────────────────────────────────────────────────────────────────
+
+export interface ListResult {
+  notifications: Notification[];
+  totalItems: number;
+}
+
+/**
+ * Returns a paginated list of notifications for a user with total count
+ * obtained via COUNT(*) OVER() in a single round-trip.
+ */
+export async function listNotificationsForUser(
+  userId: number,
+  query: NotificationQuery,
+): Promise<ListResult> {
+  if (!Number.isFinite(userId) || userId <= 0) {
+    throw new NotificationError(
+      "INVALID_USER",
+      "User id must be a positive integer",
+    );
+  }
+
+  const { page, limit, type, unreadOnly } = query;
+  const offset = (page - 1) * limit;
+
+  if (offset > NOTIFICATION_MAX_OFFSET) {
+    throw new NotificationError(
+      "PAGE_TOO_LARGE",
+      `Offset ${offset} exceeds maximum allowed ${NOTIFICATION_MAX_OFFSET}`,
+    );
+  }
+
+  // Build query with plain interpolations (no nested sql`` fragments) so
+  // the vitest mock (plain vi.fn()) correctly returns queued responses.
   let rows: Record<string, unknown>[];
 
-  if (filter.isRead !== undefined) {
+  if (type !== null && unreadOnly) {
     rows = (await sql`
-      SELECT * FROM notifications
-      WHERE  user_id = ${filter.userId}
-        AND  is_read = ${filter.isRead}
+      SELECT *, COUNT(*) OVER() AS total_count
+      FROM notifications
+      WHERE user_id = ${userId}
+        AND event_type = ${type}
+        AND is_read = FALSE
       ORDER BY created_at DESC
-      LIMIT  ${limit}
+      LIMIT ${limit}
+      OFFSET ${offset}
+    `) as Record<string, unknown>[];
+  } else if (type !== null) {
+    rows = (await sql`
+      SELECT *, COUNT(*) OVER() AS total_count
+      FROM notifications
+      WHERE user_id = ${userId}
+        AND event_type = ${type}
+      ORDER BY created_at DESC
+      LIMIT ${limit}
+      OFFSET ${offset}
+    `) as Record<string, unknown>[];
+  } else if (unreadOnly) {
+    rows = (await sql`
+      SELECT *, COUNT(*) OVER() AS total_count
+      FROM notifications
+      WHERE user_id = ${userId}
+        AND is_read = FALSE
+      ORDER BY created_at DESC
+      LIMIT ${limit}
       OFFSET ${offset}
     `) as Record<string, unknown>[];
   } else {
     rows = (await sql`
-      SELECT * FROM notifications
-      WHERE  user_id = ${filter.userId}
+      SELECT *, COUNT(*) OVER() AS total_count
+      FROM notifications
+      WHERE user_id = ${userId}
       ORDER BY created_at DESC
-      LIMIT  ${limit}
+      LIMIT ${limit}
       OFFSET ${offset}
     `) as Record<string, unknown>[];
   }
 
-  return rows.map(rowToNotification);
+  if (!rows || rows.length === 0) {
+    return { notifications: [], totalItems: 0 };
+  }
+
+  const totalItems = Number(rows[0].total_count ?? 0);
+  const notifications = rows.map(mapNotificationRow);
+
+  return { notifications, totalItems };
+}
+
+// ─── Mark single read ──────────────────────────────────────────────────────
+
+export interface MarkReadResult {
+  notification: Notification | null;
+}
+
+/**
+ * Marks a single notification as read. The update is scoped to the
+ * owner so the caller cannot mark another user's notification.
+ * Returns { notification: null } when the notification doesn't exist
+ * or belongs to someone else.
+ */
+export async function markNotificationRead(
+  notificationId: number,
+  userId: number,
+): Promise<MarkReadResult> {
+  const rows = (await sql`
+    UPDATE notifications
+    SET is_read = TRUE
+    WHERE id = ${notificationId}
+      AND user_id = ${userId}
+    RETURNING *
+  `) as Record<string, unknown>[];
+
+  if (rows.length === 0) {
+    return { notification: null };
+  }
+
+  return { notification: mapNotificationRow(rows[0]) };
+}
+
+// ─── Mark all read ─────────────────────────────────────────────────────────
+
+export interface MarkAllReadResult {
+  updatedCount: number;
 }
 
+/**
+ * Marks every unread notification for the given user as read in a
+ * single CTE so updatedCount reflects only the rows flipped by this
+ * particular call.
+ */
+export async function markAllNotificationsRead(
+  userId: number,
+): Promise<MarkAllReadResult> {
+  const rows = (await sql`
+    WITH updated AS (
+      UPDATE notifications
+      SET is_read = TRUE
+      WHERE user_id = ${userId}
+        AND is_read = FALSE
+      RETURNING id
+    )
+    SELECT COUNT(*)::int AS updated_count FROM updated
+  `) as Array<{ updated_count: number }>;
+
+  return { updatedCount: rows[0]?.updated_count ?? 0 };
+}
+
+// ─── Count ─────────────────────────────────────────────────────────────────
+
 /**
  * Returns the total count of notifications for a user, optionally filtered
  * by read status. Useful for pagination info and unread badge counts.
  */
 export async function countNotifications(
-  userId: string,
+  userId: string | number,
   isRead?: boolean,
 ): Promise<number> {
   let rows: Record<string, unknown>[];
@@ -119,18 +483,57 @@ export async function countNotifications(
   return Number(rows[0]?.count ?? 0);
 }
 
-/**
- * Returns the count of unread notifications for a user. Commonly used
- * for the badge count on the notification bell icon.
- */
-export async function getUnreadCount(userId: string): Promise<number> {
-  return countNotifications(userId, false);
+// ─── Legacy-compatible exports (used by existing main route.ts) ─────────────
+
+export interface ListNotificationsFilter {
+  userId: string;
+  isRead?: boolean;
+  limit?: number;
+  offset?: number;
+}
+
+function rowToNotification(row: Record<string, unknown>): Notification {
+  return mapNotificationRow(row);
+}
+
+export async function listNotifications(
+  filter: ListNotificationsFilter,
+): Promise<Notification[]> {
+  const limit = Math.min(filter.limit ?? 20, 100);
+  const offset = filter.offset ?? 0;
+
+  let rows: Record<string, unknown>[];
+
+  if (filter.isRead !== undefined) {
+    rows = (await sql`
+      SELECT * FROM notifications
+      WHERE  user_id = ${filter.userId}
+        AND  is_read = ${filter.isRead}
+      ORDER BY created_at DESC
+      LIMIT  ${limit}
+      OFFSET ${offset}
+    `) as Record<string, unknown>[];
+  } else {
+    rows = (await sql`
+      SELECT * FROM notifications
+      WHERE  user_id = ${filter.userId}
+      ORDER BY created_at DESC
+      LIMIT  ${limit}
+      OFFSET ${offset}
+    `) as Record<string, unknown>[];
+  }
+
+  return rows.map(rowToNotification);
+}
+
+export async function getUnreadCount(userId: string | number): Promise<number> {
+  const rows = (await sql`
+    SELECT COUNT(*) as unread FROM notifications
+    WHERE user_id = ${userId} AND is_read = FALSE
+  `) as Array<{ unread: number }>;
+  return Number(rows[0]?.unread ?? 0);
 }
 
-/**
- * Marks a single notification as read by ID. Returns true if the notification
- * existed, false otherwise.
- */
 export async function markAsRead(notificationId: string): Promise<boolean> {
   const rows = (await sql`
     UPDATE notifications
@@ -142,10 +545,6 @@ export async function markAsRead(notificationId: string): Promise<boolean> {
   return rows.length > 0;
 }
 
-/**
- * Marks a single notification as unread by ID. Returns true if the notification
- * existed, false otherwise.
- */
 export async function markAsUnread(notificationId: string): Promise<boolean> {
   const rows = (await sql`
     UPDATE notifications
@@ -157,10 +556,6 @@ export async function markAsUnread(notificationId: string): Promise<boolean> {
   return rows.length > 0;
 }
 
-/**
- * Marks all notifications for a user as read. Returns the count of
- * notifications updated.
- */
 export async function markAllAsRead(userId: string): Promise<number> {
   const rows = (await sql`
     UPDATE notifications
@@ -172,10 +567,6 @@ export async function markAllAsRead(userId: string): Promise<number> {
   return rows.length;
 }
 
-/**
- * Deletes a notification by ID. Returns true if the notification existed,
- * false otherwise.
- */
 export async function deleteNotification(
   notificationId: string,
 ): Promise<boolean> {
@@ -188,26 +579,7 @@ export async function deleteNotification(
   return rows.length > 0;
 }
 
-/**
- * Creates a notification for a user. Used internally by the system.
- * (This function is already implemented in scripts/worker.ts but included
- * here for completeness and to have a single source of truth for the service.)
- */
-export async function createNotification(
-  userId: string,
-  type: NotificationType,
-  payload: Record<string, unknown> = {},
-): Promise<Notification> {
-  const rows = (await sql`
-    INSERT INTO notifications (user_id, type, payload)
-    VALUES (${userId}, ${type}, ${JSON.stringify(payload)})
-    RETURNING *
-  `) as Record<string, unknown>[];
-
-  return rowToNotification(rows[0] as Record<string, unknown>);
-}
-
-// --- Message formatting ----------------------------------------------------
+// ─── Notification content builders ─────────────────────────────────────────
 
 export interface NotificationContent {
   title: string;
@@ -219,8 +591,6 @@ function str(payload: Record<string, unknown>, key: string): string | null {
   return typeof value === "string" && value.length > 0 ? value : null;
 }
 
-// Every NotificationType must have an entry, so adding a new event type
-// forces a template to be written for it.
 const CONTENT_BUILDERS: Record<
   NotificationType,
   (payload: Record<string, unknown>) => NotificationContent
@@ -286,7 +656,7 @@ export function buildNotificationContent(
   return CONTENT_BUILDERS[type](payload);
 }
 
-// --- Unified dispatch (in-app + email) -------------------------------------
+// ─── Unified dispatch (in-app + email) ─────────────────────────────────────
 
 export interface DispatchChannels {
   inApp?: boolean;
@@ -331,7 +701,15 @@ export async function dispatchNotification(
 
   if (wantInApp) {
     try {
-      result.notification = await createNotification(userId, type, payload);
+      const content = buildNotificationContent(type, payload);
+      result.notification = await createNotification({
+        userId: Number(userId),
+        title: content.title,
+        message: content.body,
+        type: "info",
+        eventType: type,
+        payload,
+      });
       result.inApp = true;
     } catch (err) {
       console.error(
@@ -362,3 +740,124 @@ export async function dispatchNotification(
 
   return result;
 }
+
+// ─── Event-creation helpers ────────────────────────────────────────────────
+
+function toPayload(obj: Record<string, unknown>): Record<string, unknown> {
+  return obj;
+}
+
+/**
+ * Notify both parties when a contract is created.
+ */
+export async function notifyContractCreated(
+  clientId: number,
+  freelancerId: number,
+  contractId: number,
+  contractName: string,
+): Promise<void> {
+  const payload = toPayload({ contractId, contractName });
+  await createNotification({
+    userId: clientId,
+    title: "Contract created",
+    message: `Contract "${contractName}" has been created.`,
+    type: "success",
+    eventType: "contract_created",
+    payload,
+  });
+  await createNotification({
+    userId: freelancerId,
+    title: "New contract",
+    message: `You have been assigned to "${contractName}".`,
+    type: "success",
+    eventType: "contract_created",
+    payload,
+  });
+}
+
+/**
+ * Notify the client when a freelancer submits a milestone.
+ */
+export async function notifyMilestoneSubmitted(
+  clientId: number,
+  milestoneId: number,
+  milestoneName: string,
+): Promise<void> {
+  await createNotification({
+    userId: clientId,
+    title: "Milestone submitted",
+    message: `Milestone "${milestoneName}" (#${milestoneId}) was submitted for your review.`,
+    type: "success",
+    eventType: "milestone_submitted",
+    payload: toPayload({ milestoneId, milestoneName }),
+  });
+}
+
+/**
+ * Notify the freelancer when a milestone is approved.
+ */
+export async function notifyMilestoneApproved(
+  freelancerId: number,
+  milestoneId: number,
+  milestoneName: string,
+): Promise<void> {
+  await createNotification({
+    userId: freelancerId,
+    title: "Milestone approved",
+    message: `Milestone "${milestoneName}" (#${milestoneId}) was approved.`,
+    type: "success",
+    eventType: "milestone_approved",
+    payload: toPayload({ milestoneId, milestoneName }),
+  });
+}
+
+/**
+ * Notify the client (always) and optionally the freelancer when escrow
+ * funds are released.
+ */
+export async function notifyEscrowReleased(
+  clientId: number,
+  freelancerId: number | null,
+  milestoneId: number,
+  amount: string,
+  currency: string,
+): Promise<void> {
+  const payload = toPayload({ milestoneId, amount, currency });
+  // Always notify client
+  await createNotification({
+    userId: clientId,
+    title: "Escrow released",
+    message: `${amount} ${currency} was released for milestone #${milestoneId}.`,
+    type: "success",
+    eventType: "escrow_released",
+    payload,
+  });
+  if (freelancerId !== null) {
+    await createNotification({
+      userId: freelancerId,
+      title: "Funds received",
+      message: `You received ${amount} ${currency} for milestone #${milestoneId}.`,
+      type: "success",
+      eventType: "escrow_released",
+      payload,
+    });
+  }
+}
+
+/**
+ * Notify when a dispute is opened on a contract.
+ */
+export async function notifyDisputeCreated(
+  recipientId: number,
+  disputeId: number,
+  contractId: number,
+): Promise<void> {
+  await createNotification({
+    userId: recipientId,
+    title: "Dispute opened",
+    message: `A dispute (#${disputeId}) has been opened on your contract #${contractId}.`,
+    type: "warning",
+    eventType: "dispute_created",
+    payload: toPayload({ disputeId, contractId }),
+  });
+}
diff --git a/scripts/011-notification-service.sql b/scripts/011-notification-service.sql
new file mode 100644
index 0000000..03dc328
--- /dev/null
+++ b/scripts/011-notification-service.sql
@@ -0,0 +1,96 @@
+-- 011-notification-service.sql
+--
+-- Notification service enrichment for TaskChain issue #122
+-- (Notification Service Backend).
+--
+-- The original notifications table (scripts/005-notifications.sql) is a
+-- lightweight schema with `title`, `message`, `type`, `is_read`. This
+-- migration adds the columns that the notification-service helper
+-- (lib/notifications.ts) needs without breaking existing writes from
+-- scripts/worker.ts.
+--
+-- All statements are idempotent (`ADD COLUMN IF NOT EXISTS`,
+-- `CREATE INDEX IF NOT EXISTS`) so re-running this migration on an
+-- environment that already applied it is a no-op.
+
+-- Ensure the table exists with the operational shape (integer ids, the
+-- shape that scripts/worker.ts writes against). Safe no-op when the table
+-- was created by scripts/005-notifications.sql.
+CREATE TABLE IF NOT EXISTS notifications (
+  id              SERIAL PRIMARY KEY,
+  user_id         INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+  title           VARCHAR(255) NOT NULL,
+  message         TEXT NOT NULL,
+  type            VARCHAR(50) DEFAULT 'info',
+  is_read         BOOLEAN NOT NULL DEFAULT FALSE,
+  created_at      TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+-- 1) Event-type channel: replaces the free-form `type` column with a
+--    checked enum so callers cannot invent arbitrary strings.
+ALTER TABLE notifications
+  ADD COLUMN IF NOT EXISTS event_type VARCHAR(64);
+
+-- Backfill: existing rows default to a value present in the CHECK list
+-- below. We map any unknown legacy `type` to 'info' so the ADD CONSTRAINT
+-- step cannot fail on a stray value (e.g. a hand-written 'banana').
+UPDATE notifications
+  SET event_type = CASE
+    WHEN type IN ('info', 'success', 'warning') THEN type
+    ELSE 'info'
+  END
+  WHERE event_type IS NULL;
+
+-- Constrain the column now that every row has a value. The CHECK is
+-- tolerant of legacy `info`/`success`/`warning` types the worker writes.
+DO $$
+BEGIN
+  IF NOT EXISTS (
+    SELECT 1 FROM pg_constraint WHERE conname = 'notifications_event_type_check'
+  ) THEN
+    ALTER TABLE notifications
+      ADD CONSTRAINT notifications_event_type_check
+      CHECK (
+        event_type IN (
+          'info', 'success', 'warning',
+          'contract_created',
+          'milestone_submitted',
+          'milestone_approved',
+          'escrow_released',
+          'escrow_refunded',
+          'dispute_created',
+          'dispute_resolved'
+        )
+      );
+  END IF;
+END $$;
+
+-- 2) Event-specific structured payload. Stored as JSONB so callers can
+--    do attribute-level filtering ("show all notifications about job #42")
+--    on the frontend without a schema change.
+ALTER TABLE notifications
+  ADD COLUMN IF NOT EXISTS payload JSONB NOT NULL DEFAULT '{}'::jsonb;
+
+-- 3) Delivery channel + timestamp. delivery_status lets us distinguish
+--    "created but not yet fanned out" from "delivered (or attempted)".
+ALTER TABLE notifications
+  ADD COLUMN IF NOT EXISTS channel VARCHAR(32) NOT NULL DEFAULT 'in_app';
+
+ALTER TABLE notifications
+  ADD COLUMN IF NOT EXISTS delivered_at TIMESTAMPTZ;
+
+CREATE INDEX IF NOT EXISTS idx_notifications_user
+  ON notifications(user_id);
+
+CREATE INDEX IF NOT EXISTS idx_notifications_created_at
+  ON notifications(created_at);
+
+-- Composite: (user_id, is_read, created_at DESC) is exactly the shape the
+-- listNotificationsForUser query uses for `unreadOnly=true | false` lists.
+CREATE INDEX IF NOT EXISTS idx_notifications_user_unread
+  ON notifications(user_id, is_read, created_at DESC);
+
+-- Composite: (user_id, event_type, created_at DESC) powers the `?type=`
+-- filter on the GET endpoint.
+CREATE INDEX IF NOT EXISTS idx_notifications_user_event
+  ON notifications(user_id, event_type, created_at DESC);
diff --git a/scripts/README.md b/scripts/README.md
index b60d341..5fd8c9c 100644
--- a/scripts/README.md
+++ b/scripts/README.md
@@ -106,6 +106,7 @@ CONFIRM=true npx tsx scripts/deploy-mainnet.ts
 - `006-rate-limits.sql` - Rate limiting table
 - `009-reviews-schema.sql` - Reviews table for client/freelancer ratings
 - `010-freelancer-discovery-indexes.sql` - GIN/B-tree indexes that power GET /api/freelancers (task #121)
+- `011-notification-service.sql` - Enriches the `notifications` table for the notification service backend (task #122): `event_type` CHECK constraint, `payload JSONB`, `channel`, `delivered_at`, plus `(user_id, is_read, created_at)` and `(user_id, event_type, created_at)` indexes
 
 ### Security Tables
 - `007-fail-safe.sql` - Critical operations table for fail-safe system