Phase 3.3: topic sensitivity + Phase 3.4: auth foundation

Phase 3.3 — Session Topic Sensitivity (complete): - Backend: topic_sensitivity column + session_sensitivity_log, sensitivity service (flip with one-way-latch + audit), PATCH /api/shared/chat/sessions/:id/topic, topic carried in pairing + extension WS payloads, CC filter + sensitive stats + per-mitra sensitive columns on activity page - client_app: TopicSelectionBottomSheet before pricing, topic flows through pairing request, silent WS handler for session_topic_updated - mitra_app: SensitivityBadge + SensitivityTheme + sensitivityConfigProvider, overlay badge + yellow accent, chat screen app-bar toggle with configurable confirmation + latch, extension card shows current flag, history + transcript yellow theme - control_center: Sensitivitas Topik settings section, topic filter + column with inline audit log, sensitive stats dashboard card, mitra activity sensitive columns with QC flag Phase 3.4 — Self-Managed Auth (foundation only): - Migration: auth_sessions + otp_requests tables, social identity columns on customers, password_hash + lockout on control_center_users, OTP + CC lockout app_config keys - New services: password (bcrypt + complexity), token (JWT HS256 + refresh rotation, session_id claim pre-wires future Valkey revocation), social-identity (Google + Apple JWKS), OTP (Fazpass stub — real API TBD) - Constants: AuthProvider + OtpChannel - Middleware, auth route rewrites, WS auth update, Firebase → FCM isolation still pending (next chunk); Fazpass docs + Apple Developer setup still required before E2E testing Docs: - requirement/phase3.3.md, phase3.3-plan.md, phase3.3-testing.md - requirement/phase3.4.md, phase3.4-plan.md Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-24 10:15:12 +08:00
parent 97d50a8e08
commit 780cade3db
44 changed files with 3834 additions and 103 deletions
--- a/backend/src/constants.js
+++ b/backend/src/constants.js
@@ -59,6 +59,27 @@ export const EndedBy = Object.freeze({
  MITRA: 'mitra',
 })

+// Session topic sensitivity
+export const TopicSensitivity = Object.freeze({
+  REGULAR: 'regular',
+  SENSITIVE: 'sensitive',
+})
+
+// Auth provider used to establish a session
+export const AuthProvider = Object.freeze({
+  ANONYMOUS: 'anonymous',
+  PHONE: 'phone',
+  GOOGLE: 'google',
+  APPLE: 'apple',
+  PASSWORD: 'password',
+})
+
+// OTP delivery channel
+export const OtpChannel = Object.freeze({
+  WHATSAPP: 'whatsapp',
+  SMS: 'sms',
+})
+
 // WebSocket message types
 export const WsMessage = Object.freeze({
  // Auth
@@ -93,6 +114,9 @@ export const WsMessage = Object.freeze({
  EXTENSION_REQUEST: 'extension_request',
  EXTENSION_RESPONSE: 'extension_response',

+  // Topic sensitivity
+  SESSION_TOPIC_UPDATED: 'session_topic_updated',
+
  // Early end
  EARLY_END: 'early_end',

--- a/backend/src/db/migrate.js
+++ b/backend/src/db/migrate.js
@@ -300,6 +300,115 @@ const migrate = async () => {
    ON chat_request_notifications (mitra_id, notified_at)
  `

+  // --- Phase 3.3: Session Topic Sensitivity ---
+
+  await sql`
+    ALTER TABLE chat_sessions
+    ADD COLUMN IF NOT EXISTS topic_sensitivity VARCHAR(16) NOT NULL DEFAULT 'regular'
+  `
+
+  await sql`
+    CREATE INDEX IF NOT EXISTS idx_chat_sessions_topic_sensitivity
+    ON chat_sessions (topic_sensitivity)
+  `
+
+  await sql`
+    CREATE TABLE IF NOT EXISTS session_sensitivity_log (
+      id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+      session_id UUID NOT NULL REFERENCES chat_sessions(id) ON DELETE CASCADE,
+      changed_by_mitra_id UUID NOT NULL REFERENCES mitras(id),
+      from_value VARCHAR(16) NOT NULL,
+      to_value VARCHAR(16) NOT NULL,
+      created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+    )
+  `
+
+  await sql`
+    CREATE INDEX IF NOT EXISTS idx_session_sensitivity_log_session
+    ON session_sensitivity_log (session_id)
+  `
+
+  await sql`
+    INSERT INTO app_config (key, value)
+    VALUES ('sensitive_flip_confirmation_enabled', '{"value": true}')
+    ON CONFLICT (key) DO NOTHING
+  `
+
+  await sql`
+    INSERT INTO app_config (key, value)
+    VALUES ('sensitive_flag_one_way_latch', '{"value": false}')
+    ON CONFLICT (key) DO NOTHING
+  `
+
+  // --- Phase 3.4: Self-Managed Auth ---
+
+  // Customers: add social identity columns
+  await sql`
+    ALTER TABLE customers
+      ADD COLUMN IF NOT EXISTS email VARCHAR(255),
+      ADD COLUMN IF NOT EXISTS google_sub VARCHAR(255),
+      ADD COLUMN IF NOT EXISTS apple_sub VARCHAR(255)
+  `
+  await sql`CREATE UNIQUE INDEX IF NOT EXISTS idx_customers_google_sub ON customers (google_sub) WHERE google_sub IS NOT NULL`
+  await sql`CREATE UNIQUE INDEX IF NOT EXISTS idx_customers_apple_sub ON customers (apple_sub) WHERE apple_sub IS NOT NULL`
+
+  // Control center users: password-based auth columns
+  // firebase_uid stays for backward compat during migration; will be dropped in a later cleanup migration
+  await sql`ALTER TABLE control_center_users ALTER COLUMN firebase_uid DROP NOT NULL`
+  await sql`
+    ALTER TABLE control_center_users
+      ADD COLUMN IF NOT EXISTS password_hash VARCHAR(60) NOT NULL DEFAULT '',
+      ADD COLUMN IF NOT EXISTS failed_login_count INT NOT NULL DEFAULT 0,
+      ADD COLUMN IF NOT EXISTS lockout_until TIMESTAMPTZ
+  `
+
+  // Auth sessions (refresh tokens + multi-device)
+  await sql`
+    CREATE TABLE IF NOT EXISTS auth_sessions (
+      id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+      user_type VARCHAR(16) NOT NULL,
+      user_id UUID NOT NULL,
+      refresh_token_hash VARCHAR(60) NOT NULL,
+      device_info JSONB,
+      created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+      last_used_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+      expires_at TIMESTAMPTZ NOT NULL,
+      revoked_at TIMESTAMPTZ
+    )
+  `
+  await sql`CREATE INDEX IF NOT EXISTS idx_auth_sessions_user ON auth_sessions (user_type, user_id)`
+  await sql`CREATE INDEX IF NOT EXISTS idx_auth_sessions_expires ON auth_sessions (expires_at) WHERE revoked_at IS NULL`
+
+  // OTP requests (Fazpass reference + rate-limit tracking)
+  await sql`
+    CREATE TABLE IF NOT EXISTS otp_requests (
+      id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+      phone VARCHAR(20) NOT NULL,
+      ip_address VARCHAR(45),
+      user_type VARCHAR(16) NOT NULL,
+      fazpass_reference VARCHAR(255),
+      channel VARCHAR(16),
+      attempts INT NOT NULL DEFAULT 0,
+      used_at TIMESTAMPTZ,
+      created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+      expires_at TIMESTAMPTZ NOT NULL
+    )
+  `
+  await sql`CREATE INDEX IF NOT EXISTS idx_otp_requests_phone_created ON otp_requests (phone, created_at)`
+  await sql`CREATE INDEX IF NOT EXISTS idx_otp_requests_ip_created ON otp_requests (ip_address, created_at)`
+
+  // Auth-related app_config defaults
+  await sql`
+    INSERT INTO app_config (key, value) VALUES
+      ('otp_max_per_phone_per_hour', '{"value": 3}'),
+      ('otp_max_per_ip_per_hour', '{"value": 10}'),
+      ('otp_resend_cooldown_seconds', '{"value": 60}'),
+      ('otp_verify_max_attempts', '{"value": 5}'),
+      ('cc_login_max_attempts', '{"value": 5}'),
+      ('cc_login_lockout_minutes', '{"value": 15}')
+    ON CONFLICT (key) DO NOTHING
+  `
+
  console.log('Migration complete.')
  await sql.end()
 }
--- a/backend/src/routes/internal/config.routes.js
+++ b/backend/src/routes/internal/config.routes.js
@@ -7,6 +7,7 @@ import {
  getExtensionTimeoutConfig, setExtensionTimeoutConfig,
  getEarlyEndConfig, setEarlyEndConfig,
  getMitraPingConfig, setMitraPingConfig,
+  getSensitivityConfig, setSensitivityConfig,
 } from '../../services/config.service.js'

 const attachCcUser = async (request, reply) => {
@@ -125,6 +126,28 @@ export const internalConfigRoutes = async (app) => {
    return reply.send({ success: true, data: config })
  })

+  // --- Phase 3.3: Topic Sensitivity ---
+  app.get('/sensitivity', {
+    preHandler: [authenticate, attachCcUser, requirePermission('config', 'read')],
+  }, async (request, reply) => {
+    const config = await getSensitivityConfig()
+    return reply.send({ success: true, data: config })
+  })
+
+  app.patch('/sensitivity', {
+    preHandler: [authenticate, attachCcUser, requirePermission('config', 'update')],
+  }, async (request, reply) => {
+    const { flip_confirmation_enabled, one_way_latch } = request.body ?? {}
+    if (flip_confirmation_enabled !== undefined && typeof flip_confirmation_enabled !== 'boolean') {
+      return reply.code(422).send({ success: false, error: { code: 'VALIDATION_ERROR', message: 'flip_confirmation_enabled must be a boolean' } })
+    }
+    if (one_way_latch !== undefined && typeof one_way_latch !== 'boolean') {
+      return reply.code(422).send({ success: false, error: { code: 'VALIDATION_ERROR', message: 'one_way_latch must be a boolean' } })
+    }
+    const config = await setSensitivityConfig({ flip_confirmation_enabled, one_way_latch })
+    return reply.send({ success: true, data: config })
+  })
+
  // --- Price Tiers ---
  app.get('/price-tiers', {
    preHandler: [authenticate, attachCcUser, requirePermission('config', 'read')],
--- a/backend/src/routes/internal/session.routes.js
+++ b/backend/src/routes/internal/session.routes.js
@@ -1,6 +1,7 @@
 import { authenticate, requirePermission } from '../../plugins/auth.js'
 import { getCcUserByFirebaseUid } from '../../services/cc-user.service.js'
 import { listSessions, getSessionById, rerouteSession } from '../../services/session.service.js'
+import { getSessionSensitivityLog } from '../../services/sensitivity.service.js'
 import { getDashboardStats } from '../../services/dashboard.service.js'

 const attachCcUser = async (request, reply) => {
@@ -20,8 +21,13 @@ export const sessionManagementRoutes = async (app) => {
  app.get('/', {
    preHandler: [authenticate, attachCcUser, requirePermission('config', 'read')],
  }, async (request, reply) => {
-    const { page = 1, limit = 20, status } = request.query
-    const result = await listSessions({ page: Number(page), limit: Number(limit), status })
+    const { page = 1, limit = 20, status, topic_sensitivity } = request.query
+    const result = await listSessions({
+      page: Number(page),
+      limit: Number(limit),
+      status,
+      topic_sensitivity: topic_sensitivity && topic_sensitivity !== 'all' ? topic_sensitivity : undefined,
+    })
    return reply.send({ success: true, data: result })
  })

@@ -32,7 +38,8 @@ export const sessionManagementRoutes = async (app) => {
    if (!session) {
      return reply.code(404).send({ success: false, error: { code: 'NOT_FOUND', message: 'Session not found' } })
    }
-    return reply.send({ success: true, data: session })
+    const sensitivity_log = await getSessionSensitivityLog(request.params.sessionId)
+    return reply.send({ success: true, data: { ...session, sensitivity_log } })
  })

  app.post('/:sessionId/reroute', {
--- a/backend/src/routes/public/client.chat.routes.js
+++ b/backend/src/routes/public/client.chat.routes.js
@@ -4,7 +4,7 @@ import { createPairingRequest, cancelPairingRequest } from '../../services/pairi
 import { getActiveSessionByCustomer, getActiveSessionByCustomerWithUnread, endSession, getCustomerHistory } from '../../services/session.service.js'
 import { getPricingForCustomer, isValidTier, isCustomerEligibleForFreeTrial, getFreeTrial } from '../../services/pricing.service.js'
 import { requestExtension } from '../../services/extension.service.js'
-import { EndedBy } from '../../constants.js'
+import { EndedBy, TopicSensitivity } from '../../constants.js'

 const resolveCustomer = async (request, reply) => {
  const customer = await getCustomerByFirebaseUid(request.firebaseUser.uid)
@@ -25,7 +25,14 @@ export const clientChatRoutes = async (app) => {
  })

  app.post('/request', { preHandler: [authenticate, resolveCustomer] }, async (request, reply) => {
-    const { duration_minutes, price, is_free_trial } = request.body || {}
+    const { duration_minutes, price, is_free_trial, topic_sensitivity } = request.body || {}
+
+    if (topic_sensitivity !== TopicSensitivity.REGULAR && topic_sensitivity !== TopicSensitivity.SENSITIVE) {
+      return reply.code(400).send({
+        success: false,
+        error: { code: 'BAD_REQUEST', message: 'topic_sensitivity must be regular or sensitive' },
+      })
+    }

    // Validate selection
    if (is_free_trial) {
@@ -41,6 +48,7 @@ export const clientChatRoutes = async (app) => {
        duration_minutes: freeTrial.duration_minutes,
        price: 0,
        is_free_trial: true,
+        topic_sensitivity,
      })
      return reply.code(201).send({ success: true, data: session })
    }
@@ -59,7 +67,7 @@ export const clientChatRoutes = async (app) => {
      })
    }

-    const session = await createPairingRequest(request.customer.id, { duration_minutes, price, is_free_trial: false })
+    const session = await createPairingRequest(request.customer.id, { duration_minutes, price, is_free_trial: false, topic_sensitivity })
    return reply.code(201).send({ success: true, data: session })
  })

--- a/backend/src/routes/public/shared.chat.routes.js
+++ b/backend/src/routes/public/shared.chat.routes.js
@@ -4,8 +4,9 @@ import { getMitraByFirebaseUid } from '../../services/mitra.service.js'
 import { getMessages } from '../../services/chat.service.js'
 import { getSessionClosures } from '../../services/closure.service.js'
 import { registerDeviceToken } from '../../services/notification.service.js'
+import { flipSessionSensitivity } from '../../services/sensitivity.service.js'
 import { getDb } from '../../db/client.js'
-import { UserType } from '../../constants.js'
+import { TopicSensitivity, UserType } from '../../constants.js'

 const sql = getDb()

@@ -96,4 +97,35 @@ export const sharedChatRoutes = async (app) => {
    const closure = await submitClosureMessage(sessionId, request.userType, request.userId, message)
    return reply.send({ success: true, data: closure })
  })
+
+  // Mitra flips session topic sensitivity (regular <-> sensitive)
+  app.patch('/chat/sessions/:sessionId/topic', { preHandler: [authenticate, resolveUser] }, async (request, reply) => {
+    if (request.userType !== UserType.MITRA) {
+      return reply.code(403).send({
+        success: false,
+        error: { code: 'FORBIDDEN', message: 'Only mitra can change topic sensitivity' },
+      })
+    }
+    const { sessionId } = request.params
+    const { topic_sensitivity } = request.body || {}
+    if (topic_sensitivity !== TopicSensitivity.REGULAR && topic_sensitivity !== TopicSensitivity.SENSITIVE) {
+      return reply.code(400).send({
+        success: false,
+        error: { code: 'BAD_REQUEST', message: 'topic_sensitivity must be regular or sensitive' },
+      })
+    }
+    try {
+      const result = await flipSessionSensitivity({
+        sessionId,
+        mitraId: request.userId,
+        toValue: topic_sensitivity,
+      })
+      return reply.send({ success: true, data: result })
+    } catch (err) {
+      return reply.code(err.statusCode || 500).send({
+        success: false,
+        error: { code: err.code || 'INTERNAL', message: err.message },
+      })
+    }
+  })
 }
--- a/backend/src/routes/public/shared.config.routes.js
+++ b/backend/src/routes/public/shared.config.routes.js
@@ -1,9 +1,14 @@
 import { authenticate } from '../../plugins/auth.js'
-import { getAnonymityConfig } from '../../services/config.service.js'
+import { getAnonymityConfig, getSensitivityConfig } from '../../services/config.service.js'

 export const sharedConfigRoutes = async (app) => {
  app.get('/anonymity', async (request, reply) => {
    const config = await getAnonymityConfig()
    return reply.send({ success: true, data: config })
  })
+
+  app.get('/sensitivity', { preHandler: [authenticate] }, async (request, reply) => {
+    const config = await getSensitivityConfig()
+    return reply.send({ success: true, data: config })
+  })
 }
--- a/backend/src/services/config.service.js
+++ b/backend/src/services/config.service.js
@@ -128,3 +128,97 @@ export const setEarlyEndConfig = async ({ mitra_enabled, customer_enabled }) =>
  }
  return getEarlyEndConfig()
 }
+
+// --- Phase 3.3: Session Topic Sensitivity ---
+
+export const getSensitivityConfig = async () => {
+  const [confirmRow] = await sql`SELECT value FROM app_config WHERE key = 'sensitive_flip_confirmation_enabled'`
+  const [latchRow] = await sql`SELECT value FROM app_config WHERE key = 'sensitive_flag_one_way_latch'`
+  return {
+    flip_confirmation_enabled: confirmRow?.value?.value ?? true,
+    one_way_latch: latchRow?.value?.value ?? false,
+  }
+}
+
+export const setSensitivityConfig = async ({ flip_confirmation_enabled, one_way_latch }) => {
+  if (flip_confirmation_enabled !== undefined) {
+    await sql`
+      INSERT INTO app_config (key, value, updated_at)
+      VALUES ('sensitive_flip_confirmation_enabled', ${sql.json({ value: flip_confirmation_enabled })}, NOW())
+      ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = NOW()
+    `
+  }
+  if (one_way_latch !== undefined) {
+    await sql`
+      INSERT INTO app_config (key, value, updated_at)
+      VALUES ('sensitive_flag_one_way_latch', ${sql.json({ value: one_way_latch })}, NOW())
+      ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = NOW()
+    `
+  }
+  return getSensitivityConfig()
+}
+
+// --- Phase 3.4: Self-Managed Auth ---
+
+export const getOtpRateLimits = async () => {
+  const [phoneRow] = await sql`SELECT value FROM app_config WHERE key = 'otp_max_per_phone_per_hour'`
+  const [ipRow] = await sql`SELECT value FROM app_config WHERE key = 'otp_max_per_ip_per_hour'`
+  const [resendRow] = await sql`SELECT value FROM app_config WHERE key = 'otp_resend_cooldown_seconds'`
+  const [attemptsRow] = await sql`SELECT value FROM app_config WHERE key = 'otp_verify_max_attempts'`
+  return {
+    max_per_phone_per_hour: phoneRow?.value?.value ?? 3,
+    max_per_ip_per_hour: ipRow?.value?.value ?? 10,
+    resend_cooldown_seconds: resendRow?.value?.value ?? 60,
+    verify_max_attempts: attemptsRow?.value?.value ?? 5,
+  }
+}
+
+export const setOtpRateLimits = async ({
+  max_per_phone_per_hour,
+  max_per_ip_per_hour,
+  resend_cooldown_seconds,
+  verify_max_attempts,
+}) => {
+  const pairs = [
+    ['otp_max_per_phone_per_hour', max_per_phone_per_hour],
+    ['otp_max_per_ip_per_hour', max_per_ip_per_hour],
+    ['otp_resend_cooldown_seconds', resend_cooldown_seconds],
+    ['otp_verify_max_attempts', verify_max_attempts],
+  ]
+  for (const [key, value] of pairs) {
+    if (value === undefined) continue
+    await sql`
+      INSERT INTO app_config (key, value, updated_at)
+      VALUES (${key}, ${sql.json({ value })}, NOW())
+      ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = NOW()
+    `
+  }
+  return getOtpRateLimits()
+}
+
+export const getCcLoginLockoutConfig = async () => {
+  const [attemptsRow] = await sql`SELECT value FROM app_config WHERE key = 'cc_login_max_attempts'`
+  const [minutesRow] = await sql`SELECT value FROM app_config WHERE key = 'cc_login_lockout_minutes'`
+  return {
+    max_attempts: attemptsRow?.value?.value ?? 5,
+    lockout_minutes: minutesRow?.value?.value ?? 15,
+  }
+}
+
+export const setCcLoginLockoutConfig = async ({ max_attempts, lockout_minutes }) => {
+  if (max_attempts !== undefined) {
+    await sql`
+      INSERT INTO app_config (key, value, updated_at)
+      VALUES ('cc_login_max_attempts', ${sql.json({ value: max_attempts })}, NOW())
+      ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = NOW()
+    `
+  }
+  if (lockout_minutes !== undefined) {
+    await sql`
+      INSERT INTO app_config (key, value, updated_at)
+      VALUES ('cc_login_lockout_minutes', ${sql.json({ value: lockout_minutes })}, NOW())
+      ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = NOW()
+    `
+  }
+  return getCcLoginLockoutConfig()
+}
--- a/backend/src/services/dashboard.service.js
+++ b/backend/src/services/dashboard.service.js
@@ -1,13 +1,23 @@
 import { getDb } from '../db/client.js'
-import { SessionStatus } from '../constants.js'
+import { SessionStatus, TopicSensitivity } from '../constants.js'

 const sql = getDb()

 export const getDashboardStats = async () => {
-  const [[{ active_chats }], [{ online_mitras }], [{ pending_requests }]] = await Promise.all([
+  const [
+    [{ active_chats }],
+    [{ online_mitras }],
+    [{ pending_requests }],
+    [{ sensitive_total }],
+    [{ sensitive_last_30d_total }],
+    [{ sensitive_last_30d_sensitive }],
+  ] = await Promise.all([
    sql`SELECT COUNT(*) AS active_chats FROM chat_sessions WHERE status IN (${SessionStatus.ACTIVE}, ${SessionStatus.PENDING_PAYMENT})`,
    sql`SELECT COUNT(*) AS online_mitras FROM mitra_online_status WHERE is_online = true`,
    sql`SELECT COUNT(*) AS pending_requests FROM chat_sessions WHERE status IN (${SessionStatus.SEARCHING}, ${SessionStatus.PENDING_ACCEPTANCE})`,
+    sql`SELECT COUNT(*) AS sensitive_total FROM chat_sessions WHERE topic_sensitivity = ${TopicSensitivity.SENSITIVE}`,
+    sql`SELECT COUNT(*) AS sensitive_last_30d_total FROM chat_sessions WHERE created_at >= NOW() - INTERVAL '30 days'`,
+    sql`SELECT COUNT(*) AS sensitive_last_30d_sensitive FROM chat_sessions WHERE created_at >= NOW() - INTERVAL '30 days' AND topic_sensitivity = ${TopicSensitivity.SENSITIVE}`,
  ])

  const customersPerMitra = await sql`
@@ -20,10 +30,19 @@ export const getDashboardStats = async () => {
    ORDER BY active_session_count DESC
  `

+  const last30dTotal = Number(sensitive_last_30d_total)
+  const last30dSensitive = Number(sensitive_last_30d_sensitive)
+
  return {
    active_chats: Number(active_chats),
    online_mitras: Number(online_mitras),
    pending_requests: Number(pending_requests),
    customers_per_mitra: customersPerMitra,
+    sensitive: {
+      total: Number(sensitive_total),
+      last_30d_total: last30dTotal,
+      last_30d_sensitive: last30dSensitive,
+      last_30d_percent: last30dTotal > 0 ? Math.round((last30dSensitive / last30dTotal) * 1000) / 10 : 0,
+    },
  }
 }
--- a/backend/src/services/extension.service.js
+++ b/backend/src/services/extension.service.js
@@ -17,7 +17,7 @@ const getExtensionTimeout = async () => {
 export const requestExtension = async (sessionId, customerId, { duration_minutes, price }) => {
  // Verify session belongs to customer and is in an extendable state
  const [session] = await sql`
-    SELECT id, customer_id, mitra_id, status FROM chat_sessions
+    SELECT id, customer_id, mitra_id, status, topic_sensitivity FROM chat_sessions
    WHERE id = ${sessionId} AND customer_id = ${customerId}
      AND status IN (${SessionStatus.ACTIVE}, ${SessionStatus.CLOSING})
  `
@@ -35,13 +35,14 @@ export const requestExtension = async (sessionId, customerId, { duration_minutes
  // Pause the session
  await sql`UPDATE chat_sessions SET status = ${SessionStatus.EXTENDING} WHERE id = ${sessionId}`

-  // Notify mitra
+  // Notify mitra — include current topic sensitivity so UI can highlight
  sendToSessionParticipant(sessionId, UserType.MITRA, {
    type: WsMessage.EXTENSION_REQUEST,
    extension_id: extension.id,
    session_id: sessionId,
    duration_minutes,
    price,
+    topic_sensitivity: session.topic_sensitivity,
  })

  // Notify customer that chat is paused
--- a/backend/src/services/mitra-activity.service.js
+++ b/backend/src/services/mitra-activity.service.js
@@ -1,4 +1,5 @@
 import { getDb } from '../db/client.js'
+import { TopicSensitivity } from '../constants.js'

 const sql = getDb()

@@ -18,12 +19,14 @@ export const getMitraActivityLog = async ({ mitra_id, date_from, date_to, page =
    SELECT crn.id, crn.session_id, crn.mitra_id, crn.response,
      crn.notified_at, crn.responded_at, crn.active_session_count,
      m.display_name AS mitra_display_name,
+      cs.topic_sensitivity,
      CASE WHEN crn.responded_at IS NOT NULL
        THEN EXTRACT(EPOCH FROM (crn.responded_at - crn.notified_at))::int
        ELSE NULL
      END AS response_time_seconds
    FROM chat_request_notifications crn
    INNER JOIN mitras m ON m.id = crn.mitra_id
+    LEFT JOIN chat_sessions cs ON cs.id = crn.session_id
    ${where}
    ORDER BY crn.notified_at DESC
    LIMIT ${limit} OFFSET ${offset}
@@ -63,9 +66,17 @@ export const getMitraActivitySummary = async ({ mitra_id, date_from, date_to } =
          THEN EXTRACT(EPOCH FROM (crn.responded_at - crn.notified_at))
          ELSE NULL
        END
-      )::numeric(10,1) AS avg_response_time_seconds
+      )::numeric(10,1) AS avg_response_time_seconds,
+      COUNT(*) FILTER (WHERE cs.topic_sensitivity = ${TopicSensitivity.SENSITIVE})::int AS sensitive_total,
+      COUNT(*) FILTER (WHERE cs.topic_sensitivity = ${TopicSensitivity.SENSITIVE} AND crn.response = 'accepted')::int AS sensitive_accepted,
+      COUNT(*) FILTER (WHERE cs.topic_sensitivity = ${TopicSensitivity.SENSITIVE} AND crn.response = 'declined')::int AS sensitive_rejected,
+      ROUND(
+        100.0 * COUNT(*) FILTER (WHERE cs.topic_sensitivity = ${TopicSensitivity.SENSITIVE} AND crn.response = 'accepted')
+        / NULLIF(COUNT(*) FILTER (WHERE cs.topic_sensitivity = ${TopicSensitivity.SENSITIVE}), 0), 1
+      ) AS sensitive_acceptance_rate
    FROM chat_request_notifications crn
    INNER JOIN mitras m ON m.id = crn.mitra_id
+    LEFT JOIN chat_sessions cs ON cs.id = crn.session_id
    ${where}
    GROUP BY crn.mitra_id, m.display_name
    ORDER BY acceptance_rate DESC NULLS LAST
--- a/backend/src/services/otp.service.js
+++ b/backend/src/services/otp.service.js
@@ -0,0 +1,170 @@
+import crypto from 'node:crypto'
+import { getDb } from '../db/client.js'
+import { getOtpRateLimits } from './config.service.js'
+import { OtpChannel, UserType } from '../constants.js'
+
+const sql = getDb()
+
+const OTP_TTL_MINUTES = 5
+
+// -------------------------------------------------------------------
+// ⚠ Fazpass integration — STUB until real API docs are obtained.
+//
+// In production, Fazpass is the source of truth for the OTP code.
+// We will only ever handle a reference ID (string) returned by Fazpass,
+// never the raw code. For now, we generate a 6-digit code locally and
+// store its bcrypt hash in the metadata field of otp_requests via
+// fazpass_reference (reused as "<reference>:<hash>") so the stub can
+// round-trip without schema changes.
+//
+// When real docs arrive, replace fazpassSendStub + fazpassVerifyStub
+// with real HTTP calls and drop the local code generation.
+// -------------------------------------------------------------------
+
+const generate6DigitCode = () => {
+  // Avoid Math.random for OTP generation — use crypto.randomInt
+  return String(crypto.randomInt(0, 1_000_000)).padStart(6, '0')
+}
+
+const fazpassSendStub = async ({ phone, channel }) => {
+  const reference = `stub_${crypto.randomUUID()}`
+  const code = generate6DigitCode()
+  // Log the code so developers can read it during dev testing.
+  // eslint-disable-next-line no-console
+  console.log(`[OTP STUB] phone=${phone} channel=${channel} code=${code} ref=${reference}`)
+  return { reference, channel_used: channel, code } // `code` only present in stub
+}
+
+const fazpassVerifyStub = async ({ reference, code, expectedCode }) => {
+  return { valid: code === expectedCode }
+}
+// -------------------------------------------------------------------
+
+export class OtpError extends Error {
+  constructor(message, code, statusCode) {
+    super(message)
+    this.code = code
+    this.statusCode = statusCode
+  }
+}
+
+const checkRateLimits = async ({ phone, ipAddress, limits }) => {
+  // Resend cooldown
+  const [lastRow] = await sql`
+    SELECT created_at FROM otp_requests
+    WHERE phone = ${phone}
+    ORDER BY created_at DESC LIMIT 1
+  `
+  if (lastRow) {
+    const elapsed = (Date.now() - new Date(lastRow.created_at).getTime()) / 1000
+    if (elapsed < limits.resend_cooldown_seconds) {
+      throw new OtpError(
+        `Please wait ${Math.ceil(limits.resend_cooldown_seconds - elapsed)}s before requesting another OTP`,
+        'OTP_COOLDOWN', 429,
+      )
+    }
+  }
+
+  // Per-phone hourly limit
+  const [{ phone_count }] = await sql`
+    SELECT COUNT(*)::int AS phone_count FROM otp_requests
+    WHERE phone = ${phone} AND created_at >= NOW() - INTERVAL '1 hour'
+  `
+  if (phone_count >= limits.max_per_phone_per_hour) {
+    throw new OtpError('Too many OTP requests for this number', 'OTP_RATE_LIMIT_PHONE', 429)
+  }
+
+  // Per-IP hourly limit (only if ip provided)
+  if (ipAddress) {
+    const [{ ip_count }] = await sql`
+      SELECT COUNT(*)::int AS ip_count FROM otp_requests
+      WHERE ip_address = ${ipAddress} AND created_at >= NOW() - INTERVAL '1 hour'
+    `
+    if (ip_count >= limits.max_per_ip_per_hour) {
+      throw new OtpError('Too many OTP requests from this network', 'OTP_RATE_LIMIT_IP', 429)
+    }
+  }
+}
+
+/**
+ * Start an OTP flow. Returns { otp_request_id, channel_used, expires_at }.
+ * Does NOT return the code to the caller.
+ */
+export const requestOtp = async ({ phone, userType, ipAddress, channel = OtpChannel.WHATSAPP }) => {
+  if (!phone || !/^\+[1-9]\d{6,14}$/.test(phone)) {
+    throw new OtpError('Invalid phone format (E.164 expected)', 'PHONE_INVALID', 422)
+  }
+  if (userType !== UserType.CUSTOMER && userType !== UserType.MITRA) {
+    throw new OtpError('Invalid user type', 'USER_TYPE_INVALID', 400)
+  }
+
+  const limits = await getOtpRateLimits()
+  await checkRateLimits({ phone, ipAddress, limits })
+
+  const { reference, channel_used, code } = await fazpassSendStub({ phone, channel })
+
+  // Store the reference. In stub mode, we also store the expected code appended
+  // after a colon so the verify stub can compare. Real Fazpass flow will NOT store
+  // the code; Fazpass itself holds it. This line is the main place to change
+  // when switching to real Fazpass.
+  const storedReference = code ? `${reference}:${code}` : reference
+
+  const [row] = await sql`
+    INSERT INTO otp_requests (phone, ip_address, user_type, fazpass_reference, channel, expires_at)
+    VALUES (
+      ${phone}, ${ipAddress ?? null}, ${userType}, ${storedReference}, ${channel_used},
+      NOW() + (${OTP_TTL_MINUTES} || ' minutes')::interval
+    )
+    RETURNING id, expires_at
+  `
+
+  return {
+    otp_request_id: row.id,
+    channel_used,
+    expires_at: row.expires_at,
+  }
+}
+
+/**
+ * Verify an OTP code. Returns { phone, user_type } on success.
+ * Throws OtpError on failure.
+ */
+export const verifyOtp = async ({ otpRequestId, code }) => {
+  if (typeof code !== 'string' || !/^\d{4,8}$/.test(code)) {
+    throw new OtpError('Invalid code format', 'CODE_INVALID', 422)
+  }
+
+  const limits = await getOtpRateLimits()
+
+  const [row] = await sql`
+    SELECT id, phone, user_type, fazpass_reference, attempts, used_at, expires_at
+    FROM otp_requests
+    WHERE id = ${otpRequestId}
+  `
+  if (!row) {
+    throw new OtpError('OTP request not found', 'OTP_NOT_FOUND', 404)
+  }
+  if (row.used_at) {
+    throw new OtpError('OTP already used', 'OTP_USED', 409)
+  }
+  if (new Date(row.expires_at) <= new Date()) {
+    throw new OtpError('OTP expired', 'OTP_EXPIRED', 410)
+  }
+  if (row.attempts >= limits.verify_max_attempts) {
+    throw new OtpError('Too many verification attempts; request a new OTP', 'OTP_ATTEMPTS_EXCEEDED', 429)
+  }
+
+  await sql`UPDATE otp_requests SET attempts = attempts + 1 WHERE id = ${otpRequestId}`
+
+  // Stub: fazpass_reference is stored as "<ref>:<code>"
+  const [reference, expectedCode] = (row.fazpass_reference || '').split(':')
+  const { valid } = await fazpassVerifyStub({ reference, code, expectedCode })
+
+  if (!valid) {
+    throw new OtpError('Incorrect code', 'CODE_MISMATCH', 401)
+  }
+
+  await sql`UPDATE otp_requests SET used_at = NOW() WHERE id = ${otpRequestId}`
+
+  return { phone: row.phone, user_type: row.user_type }
+}
--- a/backend/src/services/pairing.service.js
+++ b/backend/src/services/pairing.service.js
@@ -4,7 +4,7 @@ import { sendToUser } from '../plugins/websocket.js'
 import { sendPushNotification } from './notification.service.js'
 import { startSessionTimer } from './session-timer.service.js'
 import { startSessionListener } from './chat-handler.service.js'
-import { UserType, SessionStatus, NotificationResponse, TransactionType, WsMessage } from '../constants.js'
+import { UserType, SessionStatus, NotificationResponse, TransactionType, WsMessage, TopicSensitivity } from '../constants.js'

 const sql = getDb()

@@ -63,7 +63,7 @@ export const findAvailableMitras = async () => {
  return mitras
 }

-export const createPairingRequest = async (customerId, { duration_minutes, price, is_free_trial } = {}) => {
+export const createPairingRequest = async (customerId, { duration_minutes, price, is_free_trial, topic_sensitivity } = {}) => {
  // Check for existing active session or request
  const [existing] = await sql`
    SELECT id, status FROM chat_sessions
@@ -83,11 +83,15 @@ export const createPairingRequest = async (customerId, { duration_minutes, price
    })
  }

-  // Create session with duration/price
+  const resolvedTopic = topic_sensitivity === TopicSensitivity.SENSITIVE
+    ? TopicSensitivity.SENSITIVE
+    : TopicSensitivity.REGULAR
+
+  // Create session with duration/price/topic
  const [session] = await sql`
-    INSERT INTO chat_sessions (customer_id, status, duration_minutes, price, is_free_trial)
-    VALUES (${customerId}, ${SessionStatus.PENDING_ACCEPTANCE}, ${duration_minutes || null}, ${price || 0}, ${is_free_trial || false})
-    RETURNING id, customer_id, status, duration_minutes, price, is_free_trial, created_at
+    INSERT INTO chat_sessions (customer_id, status, duration_minutes, price, is_free_trial, topic_sensitivity)
+    VALUES (${customerId}, ${SessionStatus.PENDING_ACCEPTANCE}, ${duration_minutes || null}, ${price || 0}, ${is_free_trial || false}, ${resolvedTopic})
+    RETURNING id, customer_id, status, duration_minutes, price, is_free_trial, topic_sensitivity, created_at
  `

  // Create notifications for all available mitras
@@ -108,6 +112,7 @@ export const createPairingRequest = async (customerId, { duration_minutes, price
      created_at: session.created_at,
      duration_minutes: session.duration_minutes,
      is_free_trial: session.is_free_trial,
+      topic_sensitivity: session.topic_sensitivity,
    })
  }

@@ -309,7 +314,7 @@ export const expirePairingRequest = async (sessionId) => {

 export const getPendingRequestsForMitra = async (mitraId) => {
  const rows = await sql`
-    SELECT cs.id AS session_id, cs.duration_minutes, cs.is_free_trial, cs.created_at
+    SELECT cs.id AS session_id, cs.duration_minutes, cs.is_free_trial, cs.topic_sensitivity, cs.created_at
    FROM chat_request_notifications crn
    JOIN chat_sessions cs ON cs.id = crn.session_id
    WHERE crn.mitra_id = ${mitraId}
@@ -322,7 +327,8 @@ export const getPendingRequestsForMitra = async (mitraId) => {

 export const getSessionStatus = async (sessionId) => {
  const [session] = await sql`
-    SELECT cs.id, cs.customer_id, cs.mitra_id, cs.status, cs.created_at, cs.paired_at, cs.ended_at, cs.ended_by,
+    SELECT cs.id, cs.customer_id, cs.mitra_id, cs.status, cs.topic_sensitivity,
+      cs.created_at, cs.paired_at, cs.ended_at, cs.ended_by,
      m.display_name AS mitra_display_name
    FROM chat_sessions cs
    LEFT JOIN mitras m ON m.id = cs.mitra_id
--- a/backend/src/services/password.service.js
+++ b/backend/src/services/password.service.js
@@ -0,0 +1,36 @@
+import bcrypt from 'bcrypt'
+
+const BCRYPT_COST = 12
+
+// Password complexity: min 8 chars, >=1 digit, >=1 upper, >=1 lower
+export const validatePasswordComplexity = (plain) => {
+  if (typeof plain !== 'string' || plain.length < 8) {
+    throw Object.assign(new Error('Password must be at least 8 characters'), {
+      code: 'PASSWORD_TOO_SHORT', statusCode: 422,
+    })
+  }
+  if (!/[0-9]/.test(plain)) {
+    throw Object.assign(new Error('Password must contain at least one digit'), {
+      code: 'PASSWORD_MISSING_DIGIT', statusCode: 422,
+    })
+  }
+  if (!/[A-Z]/.test(plain)) {
+    throw Object.assign(new Error('Password must contain at least one uppercase letter'), {
+      code: 'PASSWORD_MISSING_UPPERCASE', statusCode: 422,
+    })
+  }
+  if (!/[a-z]/.test(plain)) {
+    throw Object.assign(new Error('Password must contain at least one lowercase letter'), {
+      code: 'PASSWORD_MISSING_LOWERCASE', statusCode: 422,
+    })
+  }
+}
+
+export const hashPassword = async (plain) => {
+  return bcrypt.hash(plain, BCRYPT_COST)
+}
+
+export const verifyPassword = async (plain, hash) => {
+  if (!hash) return false
+  return bcrypt.compare(plain, hash)
+}
--- a/backend/src/services/sensitivity.service.js
+++ b/backend/src/services/sensitivity.service.js
@@ -0,0 +1,90 @@
+import { getDb } from '../db/client.js'
+import { sendToSessionParticipant } from '../plugins/websocket.js'
+import { getSensitivityConfig } from './config.service.js'
+import { SessionStatus, TopicSensitivity, UserType, WsMessage } from '../constants.js'
+
+const sql = getDb()
+
+const FLIPPABLE_STATUSES = [SessionStatus.ACTIVE, SessionStatus.EXTENDING]
+
+export const flipSessionSensitivity = async ({ sessionId, mitraId, toValue }) => {
+  if (toValue !== TopicSensitivity.REGULAR && toValue !== TopicSensitivity.SENSITIVE) {
+    throw Object.assign(new Error('Invalid topic_sensitivity value'), {
+      code: 'INVALID_TOPIC', statusCode: 400,
+    })
+  }
+
+  const [session] = await sql`
+    SELECT id, mitra_id, status, topic_sensitivity
+    FROM chat_sessions
+    WHERE id = ${sessionId}
+  `
+  if (!session) {
+    throw Object.assign(new Error('Session not found'), {
+      code: 'SESSION_NOT_FOUND', statusCode: 404,
+    })
+  }
+  if (session.mitra_id !== mitraId) {
+    throw Object.assign(new Error('Not allowed'), {
+      code: 'FORBIDDEN', statusCode: 403,
+    })
+  }
+  if (!FLIPPABLE_STATUSES.includes(session.status)) {
+    throw Object.assign(new Error('Session is not in a flippable state'), {
+      code: 'SESSION_NOT_ACTIVE', statusCode: 409,
+    })
+  }
+
+  const fromValue = session.topic_sensitivity
+  if (fromValue === toValue) {
+    return { session_id: sessionId, topic_sensitivity: fromValue, changed: false }
+  }
+
+  const { one_way_latch } = await getSensitivityConfig()
+  if (one_way_latch && fromValue === TopicSensitivity.SENSITIVE && toValue === TopicSensitivity.REGULAR) {
+    throw Object.assign(new Error('Sensitive flag is locked and cannot be reverted'), {
+      code: 'SENSITIVITY_LATCHED', statusCode: 409,
+    })
+  }
+
+  const [log] = await sql.begin(async (tx) => {
+    await tx`
+      UPDATE chat_sessions
+      SET topic_sensitivity = ${toValue}
+      WHERE id = ${sessionId}
+    `
+    return tx`
+      INSERT INTO session_sensitivity_log (session_id, changed_by_mitra_id, from_value, to_value)
+      VALUES (${sessionId}, ${mitraId}, ${fromValue}, ${toValue})
+      RETURNING id, created_at
+    `
+  })
+
+  const payload = {
+    type: WsMessage.SESSION_TOPIC_UPDATED,
+    session_id: sessionId,
+    topic_sensitivity: toValue,
+    changed_at: log.created_at,
+  }
+  sendToSessionParticipant(sessionId, UserType.CUSTOMER, payload)
+  sendToSessionParticipant(sessionId, UserType.MITRA, payload)
+
+  return {
+    session_id: sessionId,
+    topic_sensitivity: toValue,
+    changed: true,
+    changed_at: log.created_at,
+  }
+}
+
+export const getSessionSensitivityLog = async (sessionId) => {
+  const rows = await sql`
+    SELECT l.id, l.session_id, l.changed_by_mitra_id, m.display_name AS changed_by_mitra_name,
+      l.from_value, l.to_value, l.created_at
+    FROM session_sensitivity_log l
+    LEFT JOIN mitras m ON m.id = l.changed_by_mitra_id
+    WHERE l.session_id = ${sessionId}
+    ORDER BY l.created_at ASC
+  `
+  return rows
+}
--- a/backend/src/services/session.service.js
+++ b/backend/src/services/session.service.js
@@ -6,7 +6,7 @@ const sql = getDb()

 export const getActiveSessionByCustomer = async (customerId) => {
  const [session] = await sql`
-    SELECT cs.id, cs.customer_id, cs.mitra_id, cs.status, cs.created_at, cs.paired_at,
+    SELECT cs.id, cs.customer_id, cs.mitra_id, cs.status, cs.topic_sensitivity, cs.created_at, cs.paired_at,
      cs.duration_minutes, cs.price, cs.is_free_trial, cs.expires_at, cs.extended_minutes,
      m.display_name AS mitra_display_name
    FROM chat_sessions cs
@@ -20,7 +20,7 @@ export const getActiveSessionByCustomer = async (customerId) => {

 export const getActiveSessionsByMitra = async (mitraId) => {
  const sessions = await sql`
-    SELECT cs.id, cs.customer_id, cs.status, cs.created_at, cs.paired_at,
+    SELECT cs.id, cs.customer_id, cs.status, cs.topic_sensitivity, cs.created_at, cs.paired_at,
      cs.duration_minutes, cs.expires_at, cs.extended_minutes,
      c.display_name AS customer_display_name
    FROM chat_sessions cs
@@ -120,14 +120,21 @@ export const rerouteSession = async (sessionId, newMitraId) => {
  return session
 }

-export const listSessions = async ({ page = 1, limit = 20, status } = {}) => {
+export const listSessions = async ({ page = 1, limit = 20, status, topic_sensitivity } = {}) => {
  const offset = (page - 1) * limit
-  const conditions = status
-    ? sql`WHERE cs.status = ${status}`
-    : sql``
+
+  let conditions = sql``
+  if (status && topic_sensitivity) {
+    conditions = sql`WHERE cs.status = ${status} AND cs.topic_sensitivity = ${topic_sensitivity}`
+  } else if (status) {
+    conditions = sql`WHERE cs.status = ${status}`
+  } else if (topic_sensitivity) {
+    conditions = sql`WHERE cs.topic_sensitivity = ${topic_sensitivity}`
+  }

  const items = await sql`
-    SELECT cs.id, cs.customer_id, cs.mitra_id, cs.status, cs.created_at, cs.paired_at, cs.ended_at, cs.ended_by,
+    SELECT cs.id, cs.customer_id, cs.mitra_id, cs.status, cs.topic_sensitivity,
+      cs.created_at, cs.paired_at, cs.ended_at, cs.ended_by,
      c.display_name AS customer_display_name,
      m.display_name AS mitra_display_name
    FROM chat_sessions cs
@@ -143,7 +150,8 @@ export const listSessions = async ({ page = 1, limit = 20, status } = {}) => {

 export const getSessionById = async (sessionId) => {
  const [session] = await sql`
-    SELECT cs.id, cs.customer_id, cs.mitra_id, cs.status, cs.created_at, cs.paired_at, cs.ended_at, cs.ended_by,
+    SELECT cs.id, cs.customer_id, cs.mitra_id, cs.status, cs.topic_sensitivity,
+      cs.created_at, cs.paired_at, cs.ended_at, cs.ended_by,
      cs.duration_minutes, cs.price, cs.is_free_trial, cs.expires_at, cs.extended_minutes,
      c.display_name AS customer_display_name,
      m.display_name AS mitra_display_name
@@ -159,7 +167,7 @@ export const getSessionById = async (sessionId) => {

 export const getActiveSessionByCustomerWithUnread = async (customerId) => {
  const [session] = await sql`
-    SELECT cs.id, cs.customer_id, cs.mitra_id, cs.status, cs.created_at, cs.paired_at,
+    SELECT cs.id, cs.customer_id, cs.mitra_id, cs.status, cs.topic_sensitivity, cs.created_at, cs.paired_at,
      cs.duration_minutes, cs.price, cs.is_free_trial, cs.expires_at, cs.extended_minutes,
      m.display_name AS mitra_display_name,
      (SELECT COUNT(*) FROM chat_messages cm
@@ -176,7 +184,7 @@ export const getActiveSessionByCustomerWithUnread = async (customerId) => {

 export const getActiveSessionsByMitraWithUnread = async (mitraId) => {
  const sessions = await sql`
-    SELECT cs.id, cs.customer_id, cs.status, cs.created_at, cs.paired_at,
+    SELECT cs.id, cs.customer_id, cs.status, cs.topic_sensitivity, cs.created_at, cs.paired_at,
      cs.duration_minutes, cs.expires_at, cs.extended_minutes,
      c.display_name AS customer_display_name,
      (SELECT COUNT(*) FROM chat_messages cm
@@ -194,7 +202,7 @@ export const getActiveSessionsByMitraWithUnread = async (mitraId) => {
 export const getCustomerHistory = async (customerId, { page = 1, limit = 20 } = {}) => {
  const offset = (page - 1) * limit
  const items = await sql`
-    SELECT cs.id, cs.mitra_id, cs.status, cs.created_at, cs.paired_at, cs.ended_at,
+    SELECT cs.id, cs.mitra_id, cs.status, cs.topic_sensitivity, cs.created_at, cs.paired_at, cs.ended_at,
      cs.duration_minutes, cs.price, cs.is_free_trial, cs.extended_minutes,
      m.display_name AS mitra_display_name,
      (SELECT message FROM session_closures WHERE session_id = cs.id AND user_type = ${UserType.MITRA} LIMIT 1) AS mitra_closure_message,
@@ -215,7 +223,7 @@ export const getCustomerHistory = async (customerId, { page = 1, limit = 20 } =
 export const getMitraHistory = async (mitraId, { page = 1, limit = 20 } = {}) => {
  const offset = (page - 1) * limit
  const items = await sql`
-    SELECT cs.id, cs.customer_id, cs.status, cs.created_at, cs.paired_at, cs.ended_at,
+    SELECT cs.id, cs.customer_id, cs.status, cs.topic_sensitivity, cs.created_at, cs.paired_at, cs.ended_at,
      cs.duration_minutes, cs.price, cs.is_free_trial, cs.extended_minutes,
      c.display_name AS customer_display_name,
      (SELECT message FROM session_closures WHERE session_id = cs.id AND user_type = ${UserType.MITRA} LIMIT 1) AS mitra_closure_message,
--- a/backend/src/services/social-identity.service.js
+++ b/backend/src/services/social-identity.service.js
@@ -0,0 +1,90 @@
+import { OAuth2Client } from 'google-auth-library'
+import jwt from 'jsonwebtoken'
+import jwksClient from 'jwks-rsa'
+
+const getGoogleClientIds = () => {
+  const raw = process.env.GOOGLE_OAUTH_CLIENT_IDS || ''
+  return raw.split(',').map((s) => s.trim()).filter(Boolean)
+}
+
+const googleClient = new OAuth2Client()
+
+/**
+ * Verify a Google ID token against Google's JWKS.
+ * Throws on invalid; returns { sub, email, email_verified, name } on success.
+ */
+export const verifyGoogleIdToken = async (idToken) => {
+  const audience = getGoogleClientIds()
+  if (audience.length === 0) {
+    throw Object.assign(new Error('GOOGLE_OAUTH_CLIENT_IDS not configured'), {
+      code: 'OAUTH_MISCONFIGURED', statusCode: 500,
+    })
+  }
+  try {
+    const ticket = await googleClient.verifyIdToken({ idToken, audience })
+    const payload = ticket.getPayload()
+    if (!payload?.sub) {
+      throw new Error('Google token missing sub')
+    }
+    return {
+      sub: payload.sub,
+      email: payload.email,
+      email_verified: payload.email_verified === true,
+      name: payload.name,
+    }
+  } catch (err) {
+    throw Object.assign(new Error(err.message || 'Invalid Google token'), {
+      code: 'GOOGLE_TOKEN_INVALID', statusCode: 401,
+    })
+  }
+}
+
+const appleJwks = jwksClient({
+  jwksUri: 'https://appleid.apple.com/auth/keys',
+  cache: true,
+  cacheMaxAge: 24 * 60 * 60 * 1000,
+  rateLimit: true,
+})
+
+const getApplePublicKey = (kid) => new Promise((resolve, reject) => {
+  appleJwks.getSigningKey(kid, (err, key) => {
+    if (err) return reject(err)
+    resolve(key.getPublicKey())
+  })
+})
+
+/**
+ * Verify an Apple ID token against Apple's JWKS.
+ * Throws on invalid; returns { sub, email? } on success.
+ * Note: Apple only includes `email` on first sign-in; persist it then.
+ */
+export const verifyAppleIdToken = async (idToken) => {
+  const audience = process.env.APPLE_SERVICES_ID
+  if (!audience) {
+    throw Object.assign(new Error('APPLE_SERVICES_ID not configured'), {
+      code: 'OAUTH_MISCONFIGURED', statusCode: 500,
+    })
+  }
+  try {
+    const decoded = jwt.decode(idToken, { complete: true })
+    if (!decoded?.header?.kid) throw new Error('Apple token missing kid header')
+
+    const publicKey = await getApplePublicKey(decoded.header.kid)
+    const payload = jwt.verify(idToken, publicKey, {
+      algorithms: ['RS256'],
+      audience,
+      issuer: 'https://appleid.apple.com',
+    })
+    if (!payload?.sub) throw new Error('Apple token missing sub')
+
+    return {
+      sub: payload.sub,
+      email: payload.email,
+      email_verified: payload.email_verified === true || payload.email_verified === 'true',
+    }
+  } catch (err) {
+    throw Object.assign(new Error(err.message || 'Invalid Apple token'), {
+      code: 'APPLE_TOKEN_INVALID', statusCode: 401,
+    })
+  }
+}
--- a/backend/src/services/token.service.js
+++ b/backend/src/services/token.service.js
@@ -0,0 +1,178 @@
+import crypto from 'node:crypto'
+import bcrypt from 'bcrypt'
+import jwt from 'jsonwebtoken'
+import { getDb } from '../db/client.js'
+
+const sql = getDb()
+
+const REFRESH_HASH_COST = 10
+
+const getJwtSecret = () => {
+  const secret = process.env.AUTH_JWT_SECRET
+  if (!secret || secret.length < 32) {
+    throw new Error('AUTH_JWT_SECRET is missing or too short (min 32 chars)')
+  }
+  return secret
+}
+
+const getAccessTokenTtlSeconds = () => {
+  const raw = process.env.ACCESS_TOKEN_TTL_SECONDS
+  const n = raw ? parseInt(raw, 10) : 3600
+  return Number.isFinite(n) && n > 0 ? n : 3600
+}
+
+const getRefreshTokenTtlDays = () => {
+  const raw = process.env.REFRESH_TOKEN_TTL_DAYS
+  const n = raw ? parseInt(raw, 10) : 30
+  return Number.isFinite(n) && n > 0 ? n : 30
+}
+
+const generateRefreshTokenRaw = () => {
+  return crypto.randomBytes(32).toString('base64url')
+}
+
+const signAccessToken = ({ userType, userId, sessionId }) => {
+  return jwt.sign(
+    { user_type: userType, session_id: sessionId },
+    getJwtSecret(),
+    {
+      algorithm: 'HS256',
+      expiresIn: getAccessTokenTtlSeconds(),
+      subject: userId,
+    },
+  )
+}
+
+/**
+ * Verify an access token. Returns { userType, userId, sessionId } on success.
+ * Throws an error with code/statusCode on failure.
+ */
+export const verifyAccessToken = (token) => {
+  try {
+    const decoded = jwt.verify(token, getJwtSecret(), { algorithms: ['HS256'] })
+    if (!decoded || typeof decoded !== 'object') {
+      throw new Error('Malformed token')
+    }
+    return {
+      userType: decoded.user_type,
+      userId: decoded.sub,
+      sessionId: decoded.session_id,
+    }
+  } catch (err) {
+    const code = err.name === 'TokenExpiredError' ? 'TOKEN_EXPIRED' : 'TOKEN_INVALID'
+    throw Object.assign(new Error(err.message || 'Invalid token'), {
+      code, statusCode: 401,
+    })
+  }
+}
+
+/**
+ * Create a new auth_sessions row and return tokens.
+ * deviceInfo is optional JSONB: { user_agent, ip }
+ */
+export const issueTokens = async ({ userType, userId, deviceInfo }) => {
+  const refreshRaw = generateRefreshTokenRaw()
+  const refreshHash = await bcrypt.hash(refreshRaw, REFRESH_HASH_COST)
+  const ttlDays = getRefreshTokenTtlDays()
+
+  const [session] = await sql`
+    INSERT INTO auth_sessions (user_type, user_id, refresh_token_hash, device_info, expires_at)
+    VALUES (
+      ${userType}, ${userId}, ${refreshHash}, ${deviceInfo ? sql.json(deviceInfo) : null},
+      NOW() + (${ttlDays} || ' days')::interval
+    )
+    RETURNING id, expires_at
+  `
+
+  const accessToken = signAccessToken({ userType, userId, sessionId: session.id })
+  const accessTtlSeconds = getAccessTokenTtlSeconds()
+
+  return {
+    access_token: accessToken,
+    access_token_expires_in: accessTtlSeconds,
+    refresh_token: `${session.id}.${refreshRaw}`,
+    refresh_token_expires_at: session.expires_at,
+    session_id: session.id,
+  }
+}
+
+/**
+ * Validate a refresh token and rotate it. Returns new token pair.
+ * Refresh token format: "<session_id>.<raw>"
+ */
+export const refreshTokens = async ({ refreshToken, deviceInfo }) => {
+  if (typeof refreshToken !== 'string' || !refreshToken.includes('.')) {
+    throw Object.assign(new Error('Invalid refresh token'), {
+      code: 'REFRESH_INVALID', statusCode: 401,
+    })
+  }
+  const [sessionId, raw] = refreshToken.split('.', 2)
+
+  const [row] = await sql`
+    SELECT id, user_type, user_id, refresh_token_hash, expires_at, revoked_at
+    FROM auth_sessions
+    WHERE id = ${sessionId}
+  `
+  if (!row || row.revoked_at || new Date(row.expires_at) <= new Date()) {
+    throw Object.assign(new Error('Refresh token expired or revoked'), {
+      code: 'REFRESH_INVALID', statusCode: 401,
+    })
+  }
+  const ok = await bcrypt.compare(raw, row.refresh_token_hash)
+  if (!ok) {
+    throw Object.assign(new Error('Refresh token mismatch'), {
+      code: 'REFRESH_INVALID', statusCode: 401,
+    })
+  }
+
+  // Rotate: issue a new raw, update the hash, bump last_used_at and expires_at
+  const newRaw = generateRefreshTokenRaw()
+  const newHash = await bcrypt.hash(newRaw, REFRESH_HASH_COST)
+  const ttlDays = getRefreshTokenTtlDays()
+  const [updated] = await sql`
+    UPDATE auth_sessions
+    SET refresh_token_hash = ${newHash},
+        last_used_at = NOW(),
+        expires_at = NOW() + (${ttlDays} || ' days')::interval,
+        device_info = ${deviceInfo ? sql.json(deviceInfo) : row.device_info ?? null}
+    WHERE id = ${sessionId}
+    RETURNING id, user_type, user_id, expires_at
+  `
+
+  const accessToken = signAccessToken({
+    userType: updated.user_type,
+    userId: updated.user_id,
+    sessionId: updated.id,
+  })
+
+  return {
+    access_token: accessToken,
+    access_token_expires_in: getAccessTokenTtlSeconds(),
+    refresh_token: `${updated.id}.${newRaw}`,
+    refresh_token_expires_at: updated.expires_at,
+    session_id: updated.id,
+    user_type: updated.user_type,
+    user_id: updated.user_id,
+  }
+}
+
+/**
+ * Revoke a single session (logout).
+ */
+export const revokeSession = async (sessionId) => {
+  await sql`
+    UPDATE auth_sessions SET revoked_at = NOW()
+    WHERE id = ${sessionId} AND revoked_at IS NULL
+  `
+  // Future: SADD session_id into Valkey 'revoked_sessions' with TTL = remaining access token life
+}
+
+/**
+ * Revoke every session for a user (admin ban / force logout everywhere).
+ */
+export const revokeAllSessionsForUser = async ({ userType, userId }) => {
+  await sql`
+    UPDATE auth_sessions SET revoked_at = NOW()
+    WHERE user_type = ${userType} AND user_id = ${userId} AND revoked_at IS NULL
+  `
+}