Build a Data Subject Request Handler

// src/privacy/dsr.ts — Automated data subject requests with verification and audit trail
import { pool } from "../db";
import { Redis } from "ioredis";
import { randomBytes } from "node:crypto";

const redis = new Redis(process.env.REDIS_URL!);

type RequestType = "access" | "export" | "erasure" | "rectification" | "portability" | "objection";

interface DSRequest {
  id: string;
  type: RequestType;
  userId: string;
  email: string;
  status: "pending_verification" | "verified" | "in_progress" | "review" | "completed" | "rejected";
  verificationToken: string;
  verifiedAt: string | null;
  dataSources: DataSourceResult[];
  deadline: string;            // 30 days from submission
  completedAt: string | null;
  reviewedBy: string | null;
  notes: string;
  auditLog: AuditEntry[];
  createdAt: string;
}

interface DataSourceResult {
  source: string;              // "users", "orders", "analytics", "support_tickets"
  status: "pending" | "collected" | "deleted" | "error";
  recordCount: number;
  dataPath: string | null;     // path to exported file
  deletedAt: string | null;
  error: string | null;
}

interface AuditEntry {
  action: string;
  actor: string;
  timestamp: string;
  details: string;
}

const DATA_SOURCES = [
  { name: "users", table: "users", idColumn: "id", personalFields: ["email", "name", "phone", "address"] },
  { name: "orders", table: "orders", idColumn: "customer_id", personalFields: ["billing_address", "shipping_address", "email"] },
  { name: "support_tickets", table: "support_tickets", idColumn: "user_id", personalFields: ["email", "content"] },
  { name: "activity_logs", table: "activity_logs", idColumn: "user_id", personalFields: ["ip_address", "user_agent"] },
  { name: "payments", table: "payment_records", idColumn: "customer_id", personalFields: ["card_last4", "billing_email"] },
  { name: "analytics", table: "analytics_events", idColumn: "user_id", personalFields: ["ip_address", "device_id"] },
  { name: "comments", table: "comments", idColumn: "author_id", personalFields: ["author_name", "author_email"] },
  { name: "files", table: "user_files", idColumn: "owner_id", personalFields: ["filename"] },
];

// Submit a DSR
export async function submitRequest(
  email: string,
  type: RequestType,
  notes?: string
): Promise<{ requestId: string; verificationSent: boolean }> {
  const { rows: [user] } = await pool.query("SELECT id FROM users WHERE email = $1", [email]);
  if (!user) {
    // Don't reveal whether the email exists
    return { requestId: "sent", verificationSent: true };
  }

  const id = `dsr-${Date.now().toString(36)}${randomBytes(4).toString("hex")}`;
  const verificationToken = randomBytes(32).toString("hex");
  const deadline = new Date(Date.now() + 30 * 86400000).toISOString();

  const request: DSRequest = {
    id, type, userId: user.id, email, status: "pending_verification",
    verificationToken, verifiedAt: null,
    dataSources: DATA_SOURCES.map((ds) => ({
      source: ds.name, status: "pending", recordCount: 0, dataPath: null, deletedAt: null, error: null,
    })),
    deadline, completedAt: null, reviewedBy: null,
    notes: notes || "",
    auditLog: [{ action: "submitted", actor: "user", timestamp: new Date().toISOString(), details: `${type} request submitted` }],
    createdAt: new Date().toISOString(),
  };

  await pool.query(
    `INSERT INTO dsr_requests (id, type, user_id, email, status, verification_token, data_sources, deadline, audit_log, notes, created_at)
     VALUES ($1, $2, $3, $4, 'pending_verification', $5, $6, $7, $8, $9, NOW())`,
    [id, type, user.id, email, verificationToken, JSON.stringify(request.dataSources),
     deadline, JSON.stringify(request.auditLog), notes || ""]
  );

  // Send verification email
  await redis.rpush("email:send:queue", JSON.stringify({
    to: email,
    template: "dsr_verification",
    data: { type, verifyUrl: `${process.env.APP_URL}/privacy/verify?token=${verificationToken}` },
  }));

  return { requestId: id, verificationSent: true };
}

// Verify identity and start processing
export async function verifyRequest(token: string): Promise<{ success: boolean; requestId?: string }> {
  const { rows: [req] } = await pool.query(
    "SELECT * FROM dsr_requests WHERE verification_token = $1 AND status = 'pending_verification'",
    [token]
  );
  if (!req) return { success: false };

  await pool.query(
    "UPDATE dsr_requests SET status = 'verified', verified_at = NOW() WHERE id = $1",
    [req.id]
  );

  // Start automated processing
  await processRequest(req.id);

  return { success: true, requestId: req.id };
}

// Process DSR (collect or delete data)
async function processRequest(requestId: string): Promise<void> {
  const { rows: [req] } = await pool.query("SELECT * FROM dsr_requests WHERE id = $1", [requestId]);
  const type = req.type as RequestType;

  await pool.query("UPDATE dsr_requests SET status = 'in_progress' WHERE id = $1", [requestId]);

  const dataSources: DataSourceResult[] = JSON.parse(req.data_sources);

  for (const ds of dataSources) {
    const sourceConfig = DATA_SOURCES.find((s) => s.name === ds.source);
    if (!sourceConfig) continue;

    try {
      if (type === "access" || type === "export" || type === "portability") {
        // Collect data
        const { rows, rowCount } = await pool.query(
          `SELECT * FROM ${sourceConfig.table} WHERE ${sourceConfig.idColumn} = $1`,
          [req.user_id]
        );

        ds.recordCount = rowCount || 0;

        if (rowCount && rowCount > 0) {
          // Sanitize: remove internal fields, keep only personal data
          const sanitized = rows.map((row: any) => {
            const clean: Record<string, any> = {};
            for (const field of sourceConfig.personalFields) {
              if (row[field] !== undefined) clean[field] = row[field];
            }
            clean.created_at = row.created_at;
            return clean;
          });

          // Store export file
          const exportPath = `dsr-exports/${requestId}/${ds.source}.json`;
          await pool.query(
            `INSERT INTO dsr_exports (request_id, source, data, created_at) VALUES ($1, $2, $3, NOW())`,
            [requestId, ds.source, JSON.stringify(sanitized)]
          );
          ds.dataPath = exportPath;
        }
        ds.status = "collected";
      }

      if (type === "erasure") {
        // Anonymize personal fields instead of deleting (preserve referential integrity)
        const setClauses = sourceConfig.personalFields
          .map((f) => `${f} = 'REDACTED'`)
          .join(", ");

        const result = await pool.query(
          `UPDATE ${sourceConfig.table} SET ${setClauses} WHERE ${sourceConfig.idColumn} = $1`,
          [req.user_id]
        );

        ds.recordCount = result.rowCount || 0;
        ds.status = "deleted";
        ds.deletedAt = new Date().toISOString();
      }
    } catch (err: any) {
      ds.status = "error";
      ds.error = err.message;
    }
  }

  // Update data sources
  const auditLog = JSON.parse(req.audit_log);
  auditLog.push({
    action: type === "erasure" ? "data_erased" : "data_collected",
    actor: "system",
    timestamp: new Date().toISOString(),
    details: `Processed ${dataSources.filter((d) => d.status !== "error").length}/${dataSources.length} sources`,
  });

  const hasErrors = dataSources.some((d) => d.status === "error");
  const newStatus = hasErrors ? "review" : type === "erasure" ? "completed" : "review";

  await pool.query(
    `UPDATE dsr_requests SET data_sources = $2, audit_log = $3, status = $4,
     completed_at = ${newStatus === "completed" ? "NOW()" : "NULL"} WHERE id = $1`,
    [requestId, JSON.stringify(dataSources), JSON.stringify(auditLog), newStatus]
  );

  if (newStatus === "completed") {
    await redis.rpush("email:send:queue", JSON.stringify({
      to: req.email, template: "dsr_completed",
      data: { type, requestId },
    }));
  }
}

// Check SLA compliance
export async function checkSLACompliance(): Promise<{
  total: number; overdue: number; atRisk: number;
  requests: Array<{ id: string; type: string; daysRemaining: number; status: string }>;
}> {
  const { rows } = await pool.query(
    "SELECT * FROM dsr_requests WHERE status NOT IN ('completed', 'rejected') ORDER BY deadline ASC"
  );

  const now = Date.now();
  let overdue = 0;
  let atRisk = 0;

  const requests = rows.map((r: any) => {
    const daysRemaining = Math.ceil((new Date(r.deadline).getTime() - now) / 86400000);
    if (daysRemaining < 0) overdue++;
    else if (daysRemaining < 7) atRisk++;
    return { id: r.id, type: r.type, daysRemaining, status: r.status };
  });

  return { total: rows.length, overdue, atRisk, requests };
}