Skip to main content

Immutable Audit Trail Storage Architecture

Executive Summary​

This document defines the comprehensive immutable audit trail storage system for the BIO-QMS platform, a multi-tenant biomedical/pharmaceutical Quality Management System (QMS) SaaS application subject to FDA 21 CFR Part 11, HIPAA Security Rule, and SOC 2 Type II compliance requirements.

The architecture implements a cryptographically-secured, append-only audit trail with:

  • Tamper-proof storage via PostgreSQL append-only design with database triggers preventing modification
  • Cryptographic integrity through SHA-256 hash chains linking records with external root hash verification
  • Multi-tenant isolation using PostgreSQL Row-Level Security (RLS) policies
  • Regulatory retention with automated hot/warm/cold storage tiering (7-year FDA, 6-year HIPAA, 1-year SOC 2)
  • High performance with <5ms write latency and partitioned queries for millions of records

Compliance Mapping:

RegulationControlRequirementImplementation
FDA 21 CFR Part 11§11.10(e)Audit trails for record changesHash-chained append-only audit records
HIPAA§164.312(b)Audit controlsRLS-enforced multi-tenant audit logging
SOC 2CC7.2System monitoringAutomated audit trail verification
SOC 2CC7.3Monitoring activities evaluationDaily hash chain integrity checks
SOC 2CC7.4Monitoring results remediationAutomated alerting on integrity failures

Table of Contents​

  1. Architecture Overview
  2. Append-Only Storage Design
  3. Cryptographic Hash Chain
  4. Multi-Tenant Isolation
  5. Retention & Archival Pipeline
  6. Performance Optimization
  7. API Integration
  8. Verification & Monitoring
  9. Compliance Controls
  10. Implementation Guide

1. Architecture Overview​

1.1 System Components​

1.2 Design Principles​

  1. Immutability First: All audit records are append-only; no updates or deletions permitted
  2. Cryptographic Proof: Hash chains provide mathematical proof of record integrity
  3. Zero Trust: Database-level enforcement prevents application-layer tampering
  4. Tenant Isolation: RLS policies ensure cross-tenant audit data cannot be accessed
  5. Performance at Scale: Partitioning and indexing support millions of audit records with <5ms writes
  6. Regulatory Compliance: Retention policies automatically enforced per regulation

2. Append-Only Storage Design​

2.1 Database Schema​

-- ============================================================================
-- AUDIT TRAIL TABLE - IMMUTABLE STORAGE
-- ============================================================================

CREATE TABLE audit_trail (
    -- Primary Key
    id                  BIGSERIAL PRIMARY KEY,

    -- Temporal
    timestamp           TIMESTAMPTZ NOT NULL DEFAULT NOW(),

    -- Multi-Tenant Context
    tenant_id           UUID NOT NULL REFERENCES tenants(id),
    user_id             UUID REFERENCES users(id),
    session_id          UUID,

    -- Action Context
    action_type         VARCHAR(50) NOT NULL,  -- CREATE, UPDATE, DELETE, READ, EXECUTE
    resource_type       VARCHAR(100) NOT NULL, -- document, sop, capa, deviation, etc.
    resource_id         UUID,

    -- Change Tracking (JSONB for flexible schema)
    old_value           JSONB,
    new_value           JSONB,

    -- Additional Metadata
    metadata            JSONB NOT NULL DEFAULT '{}'::jsonb,

    -- Cryptographic Hash Chain
    hash                CHAR(64) NOT NULL,           -- SHA-256 of this record
    previous_hash       CHAR(64),                     -- Links to prior record

    -- Audit Metadata
    created_at          TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    ip_address          INET,
    user_agent          TEXT,

    -- Indexing Hints
    CONSTRAINT valid_action_type CHECK (action_type IN (
        'CREATE', 'UPDATE', 'DELETE', 'READ',
        'EXECUTE', 'APPROVE', 'REJECT', 'SIGN', 'EXPORT'
    ))
) PARTITION BY RANGE (timestamp);

-- ============================================================================
-- PARTITION STRATEGY - Monthly partitions for 2 years hot storage
-- ============================================================================

CREATE TABLE audit_trail_2026_02 PARTITION OF audit_trail
    FOR VALUES FROM ('2026-02-01') TO ('2026-03-01');

CREATE TABLE audit_trail_2026_03 PARTITION OF audit_trail
    FOR VALUES FROM ('2026-03-01') TO ('2026-04-01');

-- ... (automated partition creation via cron)

-- ============================================================================
-- INDEXES FOR PERFORMANCE
-- ============================================================================

-- Primary lookup: tenant + timestamp range queries
CREATE INDEX idx_audit_tenant_timestamp
    ON audit_trail (tenant_id, timestamp DESC);

-- Resource lookup: find all changes to a specific record
CREATE INDEX idx_audit_resource
    ON audit_trail (tenant_id, resource_type, resource_id, timestamp DESC);

-- User activity tracking
CREATE INDEX idx_audit_user
    ON audit_trail (tenant_id, user_id, timestamp DESC);

-- Hash chain verification
CREATE INDEX idx_audit_hash_chain
    ON audit_trail (tenant_id, timestamp ASC)
    INCLUDE (hash, previous_hash);

-- Action type filtering
CREATE INDEX idx_audit_action_type
    ON audit_trail (tenant_id, action_type, timestamp DESC);

-- JSONB field indexes for common queries
CREATE INDEX idx_audit_metadata_gin
    ON audit_trail USING GIN (metadata jsonb_path_ops);

-- ============================================================================
-- IMMUTABILITY ENFORCEMENT - Database Triggers
-- ============================================================================

-- Prevent UPDATE operations
CREATE OR REPLACE FUNCTION prevent_audit_update()
RETURNS TRIGGER AS $$
BEGIN
    RAISE EXCEPTION 'Audit records are immutable - UPDATE operations are not permitted'
        USING ERRCODE = 'P0001',
              HINT = 'Audit trail integrity violation detected';
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER trg_prevent_audit_update
    BEFORE UPDATE ON audit_trail
    FOR EACH ROW
    EXECUTE FUNCTION prevent_audit_update();

-- Prevent DELETE operations
CREATE OR REPLACE FUNCTION prevent_audit_delete()
RETURNS TRIGGER AS $$
BEGIN
    RAISE EXCEPTION 'Audit records are immutable - DELETE operations are not permitted'
        USING ERRCODE = 'P0001',
              HINT = 'Audit trail integrity violation detected';
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER trg_prevent_audit_delete
    BEFORE DELETE ON audit_trail
    FOR EACH ROW
    EXECUTE FUNCTION prevent_audit_delete();

-- ============================================================================
-- HASH CHAIN COMPUTATION - Automatic on INSERT
-- ============================================================================

CREATE OR REPLACE FUNCTION compute_audit_hash()
RETURNS TRIGGER AS $$
DECLARE
    v_previous_hash CHAR(64);
    v_hash_input TEXT;
    v_computed_hash CHAR(64);
BEGIN
    -- Get the most recent hash for this tenant
    SELECT hash INTO v_previous_hash
    FROM audit_trail
    WHERE tenant_id = NEW.tenant_id
    ORDER BY timestamp DESC, id DESC
    LIMIT 1;

    -- If no previous hash exists, use null genesis value
    v_previous_hash := COALESCE(v_previous_hash, REPEAT('0', 64));

    -- Compute hash input: previous_hash + timestamp + tenant + user + action + resource + values
    v_hash_input := v_previous_hash ||
                    EXTRACT(EPOCH FROM NEW.timestamp)::TEXT ||
                    NEW.tenant_id::TEXT ||
                    COALESCE(NEW.user_id::TEXT, '') ||
                    NEW.action_type ||
                    NEW.resource_type ||
                    COALESCE(NEW.resource_id::TEXT, '') ||
                    COALESCE(NEW.old_value::TEXT, '') ||
                    COALESCE(NEW.new_value::TEXT, '');

    -- Compute SHA-256 hash
    v_computed_hash := encode(digest(v_hash_input, 'sha256'), 'hex');

    -- Set hash fields
    NEW.hash := v_computed_hash;
    NEW.previous_hash := v_previous_hash;

    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER trg_compute_audit_hash
    BEFORE INSERT ON audit_trail
    FOR EACH ROW
    EXECUTE FUNCTION compute_audit_hash();

2.2 TypeScript Audit Record Interface​

// ============================================================================
// AUDIT TRAIL TYPE DEFINITIONS
// ============================================================================

export enum AuditActionType {
  CREATE = 'CREATE',
  UPDATE = 'UPDATE',
  DELETE = 'DELETE',
  READ = 'READ',
  EXECUTE = 'EXECUTE',
  APPROVE = 'APPROVE',
  REJECT = 'REJECT',
  SIGN = 'SIGN',
  EXPORT = 'EXPORT',
}

export enum AuditResourceType {
  DOCUMENT = 'document',
  SOP = 'sop',
  CAPA = 'capa',
  DEVIATION = 'deviation',
  CHANGE_CONTROL = 'change_control',
  TRAINING_RECORD = 'training_record',
  VALIDATION_PROTOCOL = 'validation_protocol',
  AUDIT_FINDING = 'audit_finding',
  ELECTRONIC_SIGNATURE = 'electronic_signature',
  USER = 'user',
  TENANT = 'tenant',
  ROLE = 'role',
  PERMISSION = 'permission',
}

export interface AuditRecord {
  id: string;
  timestamp: Date;

  // Multi-tenant context
  tenantId: string;
  userId?: string;
  sessionId?: string;

  // Action context
  actionType: AuditActionType;
  resourceType: AuditResourceType;
  resourceId?: string;

  // Change tracking
  oldValue?: Record<string, any>;
  newValue?: Record<string, any>;

  // Metadata
  metadata: AuditMetadata;

  // Hash chain
  hash: string;
  previousHash?: string;

  // Audit metadata
  createdAt: Date;
  ipAddress?: string;
  userAgent?: string;
}

export interface AuditMetadata {
  // HTTP request context
  method?: string;
  endpoint?: string;
  statusCode?: number;

  // Change description
  description?: string;
  changeReason?: string;

  // Regulatory context
  validationRequired?: boolean;
  gxpImpact?: boolean;
  regulatoryJustification?: string;

  // Workflow context
  workflowStep?: string;
  approvalChain?: string[];

  // Additional fields (extensible)
  [key: string]: any;
}

// ============================================================================
// AUDIT QUERY INTERFACES
// ============================================================================

export interface AuditQueryFilters {
  tenantId: string;

  // Temporal filters
  startDate?: Date;
  endDate?: Date;

  // Context filters
  userId?: string;
  sessionId?: string;

  // Action filters
  actionTypes?: AuditActionType[];
  resourceTypes?: AuditResourceType[];
  resourceId?: string;

  // Metadata filters
  metadataQuery?: Record<string, any>;

  // Pagination
  limit?: number;
  offset?: number;
  sortOrder?: 'ASC' | 'DESC';
}

export interface AuditQueryResult {
  records: AuditRecord[];
  totalCount: number;
  hasMore: boolean;
  nextOffset?: number;
}

// ============================================================================
// HASH CHAIN VERIFICATION INTERFACES
// ============================================================================

export interface HashChainVerificationResult {
  tenantId: string;
  startTimestamp: Date;
  endTimestamp: Date;
  totalRecords: number;
  verifiedRecords: number;
  failedRecords: number;
  failures: HashChainFailure[];
  merkleRootHash: string;
  verificationTimestamp: Date;
  durationMs: number;
}

export interface HashChainFailure {
  recordId: string;
  timestamp: Date;
  expectedHash: string;
  actualHash: string;
  previousHash: string;
  reason: string;
}

3. Cryptographic Hash Chain​

3.1 Hash Chain Design​

Each audit record contains:

  1. hash: SHA-256 hash of the current record's canonical representation
  2. previous_hash: Hash of the immediately preceding record (for this tenant)

This creates a cryptographically-linked chain where:

  • Modifying any historical record breaks the chain (hashes no longer match)
  • Inserting a record out-of-sequence is detectable (timestamp/hash mismatch)
  • Deleting a record creates a gap (previous_hash points to non-existent record)

3.2 Hash Computation Algorithm​

// ============================================================================
// HASH CHAIN BUILDER
// ============================================================================

import { createHash } from 'crypto';

export class HashChainBuilder {
  /**
   * Compute SHA-256 hash for an audit record
   *
   * Hash input format (canonical representation):
   * previous_hash + timestamp + tenant_id + user_id + action_type +
   * resource_type + resource_id + old_value + new_value
   */
  static computeHash(
    previousHash: string,
    timestamp: Date,
    tenantId: string,
    userId: string | undefined,
    actionType: string,
    resourceType: string,
    resourceId: string | undefined,
    oldValue: Record<string, any> | undefined,
    newValue: Record<string, any> | undefined
  ): string {
    // Canonicalize null/undefined to empty string
    const canonicalUserId = userId ?? '';
    const canonicalResourceId = resourceId ?? '';
    const canonicalOldValue = oldValue ? JSON.stringify(this.sortKeys(oldValue)) : '';
    const canonicalNewValue = newValue ? JSON.stringify(this.sortKeys(newValue)) : '';

    // Build hash input string
    const hashInput = [
      previousHash,
      timestamp.getTime().toString(),
      tenantId,
      canonicalUserId,
      actionType,
      resourceType,
      canonicalResourceId,
      canonicalOldValue,
      canonicalNewValue,
    ].join('');

    // Compute SHA-256
    return createHash('sha256')
      .update(hashInput, 'utf8')
      .digest('hex');
  }

  /**
   * Sort object keys recursively for canonical JSON representation
   * (ensures consistent hashing regardless of property order)
   */
  private static sortKeys(obj: any): any {
    if (obj === null || typeof obj !== 'object') {
      return obj;
    }

    if (Array.isArray(obj)) {
      return obj.map(item => this.sortKeys(item));
    }

    return Object.keys(obj)
      .sort()
      .reduce((result, key) => {
        result[key] = this.sortKeys(obj[key]);
        return result;
      }, {} as any);
  }

  /**
   * Verify hash chain integrity for a sequence of records
   */
  static verifyChain(records: AuditRecord[]): HashChainVerificationResult {
    const startTime = Date.now();
    const failures: HashChainFailure[] = [];

    // Sort by timestamp ascending
    const sortedRecords = [...records].sort(
      (a, b) => a.timestamp.getTime() - b.timestamp.getTime()
    );

    for (let i = 0; i < sortedRecords.length; i++) {
      const record = sortedRecords[i];
      const expectedPreviousHash = i === 0
        ? '0'.repeat(64)  // Genesis record
        : sortedRecords[i - 1].hash;

      // Verify previous_hash linkage
      if (record.previousHash !== expectedPreviousHash) {
        failures.push({
          recordId: record.id,
          timestamp: record.timestamp,
          expectedHash: expectedPreviousHash,
          actualHash: record.previousHash ?? '',
          previousHash: record.previousHash ?? '',
          reason: 'Previous hash mismatch - chain broken',
        });
      }

      // Recompute hash and verify
      const recomputedHash = this.computeHash(
        record.previousHash ?? '0'.repeat(64),
        record.timestamp,
        record.tenantId,
        record.userId,
        record.actionType,
        record.resourceType,
        record.resourceId,
        record.oldValue,
        record.newValue
      );

      if (recomputedHash !== record.hash) {
        failures.push({
          recordId: record.id,
          timestamp: record.timestamp,
          expectedHash: recomputedHash,
          actualHash: record.hash,
          previousHash: record.previousHash ?? '',
          reason: 'Hash mismatch - record tampered',
        });
      }
    }

    return {
      tenantId: records[0]?.tenantId ?? 'unknown',
      startTimestamp: sortedRecords[0]?.timestamp ?? new Date(),
      endTimestamp: sortedRecords[sortedRecords.length - 1]?.timestamp ?? new Date(),
      totalRecords: records.length,
      verifiedRecords: records.length - failures.length,
      failedRecords: failures.length,
      failures,
      merkleRootHash: this.computeMerkleRoot(sortedRecords.map(r => r.hash)),
      verificationTimestamp: new Date(),
      durationMs: Date.now() - startTime,
    };
  }

  /**
   * Compute Merkle tree root hash for efficient batch verification
   */
  static computeMerkleRoot(hashes: string[]): string {
    if (hashes.length === 0) return '0'.repeat(64);
    if (hashes.length === 1) return hashes[0];

    // Build Merkle tree bottom-up
    let currentLevel = [...hashes];

    while (currentLevel.length > 1) {
      const nextLevel: string[] = [];

      for (let i = 0; i < currentLevel.length; i += 2) {
        const left = currentLevel[i];
        const right = i + 1 < currentLevel.length
          ? currentLevel[i + 1]
          : left; // Duplicate last hash if odd number

        const combined = createHash('sha256')
          .update(left + right, 'utf8')
          .digest('hex');

        nextLevel.push(combined);
      }

      currentLevel = nextLevel;
    }

    return currentLevel[0];
  }
}

3.3 External Root Hash Verification​

// ============================================================================
// EXTERNAL ROOT HASH STORE (Cloud KMS Signed)
// ============================================================================

import { KeyManagementServiceClient } from '@google-cloud/kms';

export class ExternalHashStore {
  private kmsClient: KeyManagementServiceClient;
  private keyName: string;

  constructor(projectId: string, locationId: string, keyRingId: string, keyId: string) {
    this.kmsClient = new KeyManagementServiceClient();
    this.keyName = this.kmsClient.cryptoKeyPath(
      projectId,
      locationId,
      keyRingId,
      keyId
    );
  }

  /**
   * Publish daily Merkle root hash to immutable external store
   * Signed with Cloud KMS asymmetric key for non-repudiation
   */
  async publishDailyRootHash(
    tenantId: string,
    date: Date,
    merkleRootHash: string,
    recordCount: number
  ): Promise<void> {
    // Create canonical message
    const message = JSON.stringify({
      tenantId,
      date: date.toISOString(),
      merkleRootHash,
      recordCount,
      timestamp: new Date().toISOString(),
    });

    // Sign with KMS
    const [signResponse] = await this.kmsClient.asymmetricSign({
      name: `${this.keyName}/cryptoKeyVersions/1`,
      digest: {
        sha256: createHash('sha256').update(message, 'utf8').digest(),
      },
    });

    const signature = signResponse.signature?.toString('base64') ?? '';

    // Store in immutable storage (Cloud Storage with object versioning)
    await this.storeInGCS(tenantId, date, message, signature);

    // Publish event for monitoring
    await this.publishEvent({
      type: 'audit.root_hash_published',
      tenantId,
      date,
      merkleRootHash,
      recordCount,
      signature,
    });
  }

  /**
   * Verify published root hash signature
   */
  async verifyRootHash(
    tenantId: string,
    date: Date
  ): Promise<{ valid: boolean; message: string }> {
    // Retrieve from GCS
    const { message, signature } = await this.retrieveFromGCS(tenantId, date);

    // Verify signature with KMS
    const [verifyResponse] = await this.kmsClient.asymmetricDecrypt({
      name: `${this.keyName}/cryptoKeyVersions/1`,
      ciphertext: Buffer.from(signature, 'base64'),
    });

    const isValid = verifyResponse.plaintext !== null;

    return {
      valid: isValid,
      message: isValid
        ? 'Root hash signature verified successfully'
        : 'Root hash signature verification failed',
    };
  }

  private async storeInGCS(
    tenantId: string,
    date: Date,
    message: string,
    signature: string
  ): Promise<void> {
    // Implementation: Upload to gs://audit-root-hashes/{tenantId}/{YYYY}/{MM}/{DD}.json
    // Object versioning enabled on bucket for immutability
  }

  private async retrieveFromGCS(
    tenantId: string,
    date: Date
  ): Promise<{ message: string; signature: string }> {
    // Implementation: Download from GCS
    throw new Error('Not implemented');
  }

  private async publishEvent(event: any): Promise<void> {
    // Implementation: Publish to Cloud Pub/Sub for monitoring
  }
}

4. Multi-Tenant Isolation​

4.1 Row-Level Security (RLS) Policies​

-- ============================================================================
-- ENABLE ROW-LEVEL SECURITY
-- ============================================================================

ALTER TABLE audit_trail ENABLE ROW LEVEL SECURITY;

-- ============================================================================
-- RLS POLICY: Users can only see their own tenant's audit records
-- ============================================================================

CREATE POLICY audit_tenant_isolation ON audit_trail
    FOR SELECT
    USING (tenant_id = current_setting('app.current_tenant_id', true)::uuid);

-- ============================================================================
-- RLS POLICY: Superadmin bypass (requires break-glass procedure)
-- ============================================================================

-- Note: Superadmin access is NOT granted by default
-- Must be explicitly enabled via break-glass procedure with separate audit trail

CREATE POLICY audit_superadmin_access ON audit_trail
    FOR SELECT
    TO superadmin_role
    USING (true);  -- Full access, but logged separately

-- ============================================================================
-- APPLICATION ROLE (non-superadmin)
-- ============================================================================

CREATE ROLE app_user;

-- Grant only INSERT permission (read via RLS, no UPDATE/DELETE)
GRANT INSERT ON audit_trail TO app_user;
GRANT SELECT ON audit_trail TO app_user;

-- ============================================================================
-- TENANT CONTEXT SETTER
-- ============================================================================

CREATE OR REPLACE FUNCTION set_tenant_context(p_tenant_id UUID)
RETURNS void AS $$
BEGIN
    -- Set session-local tenant context for RLS
    PERFORM set_config('app.current_tenant_id', p_tenant_id::text, true);

    -- Audit the context switch
    INSERT INTO tenant_context_audit (
        tenant_id,
        set_at,
        set_by
    ) VALUES (
        p_tenant_id,
        NOW(),
        current_user
    );
END;
$$ LANGUAGE plpgsql SECURITY DEFINER;

4.2 TypeScript Multi-Tenant Context Management​

// ============================================================================
// TENANT CONTEXT MANAGER
// ============================================================================

import { Pool, PoolClient } from 'pg';

export class TenantContextManager {
  constructor(private pool: Pool) {}

  /**
   * Execute a function within a tenant context
   * Ensures RLS policy is applied for all queries in the callback
   */
  async withTenantContext<T>(
    tenantId: string,
    callback: (client: PoolClient) => Promise<T>
  ): Promise<T> {
    const client = await this.pool.connect();

    try {
      // Begin transaction
      await client.query('BEGIN');

      // Set tenant context for RLS
      await client.query(
        'SET LOCAL app.current_tenant_id = $1',
        [tenantId]
      );

      // Execute callback with tenant-scoped client
      const result = await callback(client);

      // Commit transaction
      await client.query('COMMIT');

      return result;
    } catch (error) {
      // Rollback on error
      await client.query('ROLLBACK');
      throw error;
    } finally {
      // Always release client back to pool
      client.release();
    }
  }

  /**
   * Verify tenant isolation by attempting cross-tenant access
   * (For testing/validation purposes)
   */
  async verifyTenantIsolation(
    tenantA: string,
    tenantB: string
  ): Promise<{ isolated: boolean; message: string }> {
    return await this.withTenantContext(tenantA, async (client) => {
      // Attempt to query tenant B's audit records while scoped to tenant A
      const result = await client.query(
        'SELECT COUNT(*) as count FROM audit_trail WHERE tenant_id = $1',
        [tenantB]
      );

      const count = parseInt(result.rows[0].count, 10);

      return {
        isolated: count === 0,
        message: count === 0
          ? 'Tenant isolation verified - cross-tenant access blocked'
          : `Tenant isolation FAILED - ${count} records from tenant B visible to tenant A`,
      };
    });
  }
}

5. Retention & Archival Pipeline​

5.1 Storage Tier Architecture​

TierStorageRetentionAccess PatternCost
HotPostgreSQL0-90 daysReal-time queriesHigh
WarmBigQuery90 days - 2 yearsAnalytical queriesMedium
ColdGCS Archive2+ yearsCompliance retrievalLow

5.2 Archival Pipeline Implementation​

// ============================================================================
// AUDIT ARCHIVAL SERVICE
// ============================================================================

import { BigQuery } from '@google-cloud/bigquery';
import { Storage } from '@google-cloud/storage';
import { Pool } from 'pg';

export interface ArchivalPolicy {
  hotRetentionDays: number;      // 90
  warmRetentionDays: number;     // 730 (2 years)
  coldRetentionDays: number;     // 2555 (7 years for FDA)
}

export class AuditArchivalService {
  private bigquery: BigQuery;
  private storage: Storage;
  private pool: Pool;

  constructor(pool: Pool) {
    this.pool = pool;
    this.bigquery = new BigQuery();
    this.storage = new Storage();
  }

  /**
   * Archive hot storage (PostgreSQL) to warm storage (BigQuery)
   * Run daily via cron for records older than 90 days
   */
  async archiveToWarmStorage(tenantId: string): Promise<void> {
    const cutoffDate = new Date();
    cutoffDate.setDate(cutoffDate.getDate() - 90);

    // Extract records older than 90 days
    const records = await this.extractHotRecords(tenantId, cutoffDate);

    if (records.length === 0) {
      console.log(`No records to archive for tenant ${tenantId}`);
      return;
    }

    // Verify hash chain before archival
    const verification = HashChainBuilder.verifyChain(records);
    if (verification.failedRecords > 0) {
      throw new Error(
        `Hash chain verification failed - cannot archive tampered records. ` +
        `Failures: ${verification.failedRecords}`
      );
    }

    // Load into BigQuery
    await this.loadToBigQuery(tenantId, records);

    // Verify BigQuery load
    const loadedCount = await this.countBigQueryRecords(
      tenantId,
      records[0].timestamp,
      records[records.length - 1].timestamp
    );

    if (loadedCount !== records.length) {
      throw new Error(
        `BigQuery load verification failed. ` +
        `Expected ${records.length}, got ${loadedCount}`
      );
    }

    // Delete from PostgreSQL (after successful verification)
    await this.deleteHotRecords(tenantId, cutoffDate);

    // Log archival event
    await this.logArchivalEvent({
      tenantId,
      fromTier: 'hot',
      toTier: 'warm',
      recordCount: records.length,
      startDate: records[0].timestamp,
      endDate: records[records.length - 1].timestamp,
      merkleRootHash: verification.merkleRootHash,
    });
  }

  /**
   * Archive warm storage (BigQuery) to cold storage (GCS Archive)
   * Run weekly via cron for records older than 2 years
   */
  async archiveToColdStorage(tenantId: string): Promise<void> {
    const cutoffDate = new Date();
    cutoffDate.setDate(cutoffDate.getDate() - 730); // 2 years

    // Extract from BigQuery as Parquet
    const exportUri = await this.exportBigQueryToParquet(tenantId, cutoffDate);

    // Move to Archive storage class
    await this.moveToArchiveClass(exportUri);

    // Verify cold storage integrity
    const verified = await this.verifyColdStorage(exportUri);
    if (!verified) {
      throw new Error('Cold storage verification failed');
    }

    // Delete from BigQuery
    await this.deleteBigQueryRecords(tenantId, cutoffDate);

    // Log archival event
    await this.logArchivalEvent({
      tenantId,
      fromTier: 'warm',
      toTier: 'cold',
      exportUri,
      cutoffDate,
    });
  }

  /**
   * Restore archived records for compliance retrieval
   */
  async restoreFromArchive(
    tenantId: string,
    startDate: Date,
    endDate: Date
  ): Promise<AuditRecord[]> {
    // Determine which tier(s) to query
    const now = new Date();
    const warmCutoff = new Date(now.getTime() - 90 * 24 * 60 * 60 * 1000);
    const coldCutoff = new Date(now.getTime() - 730 * 24 * 60 * 60 * 1000);

    const records: AuditRecord[] = [];

    // Query hot storage (PostgreSQL)
    if (endDate >= warmCutoff) {
      const hotRecords = await this.queryHotStorage(tenantId, startDate, endDate);
      records.push(...hotRecords);
    }

    // Query warm storage (BigQuery)
    if (startDate < warmCutoff && endDate >= coldCutoff) {
      const warmRecords = await this.queryWarmStorage(tenantId, startDate, endDate);
      records.push(...warmRecords);
    }

    // Query cold storage (GCS Archive)
    if (startDate < coldCutoff) {
      const coldRecords = await this.queryColdStorage(tenantId, startDate, endDate);
      records.push(...coldRecords);
    }

    // Sort by timestamp
    return records.sort((a, b) => a.timestamp.getTime() - b.timestamp.getTime());
  }

  // ========================================================================
  // PRIVATE HELPER METHODS
  // ========================================================================

  private async extractHotRecords(
    tenantId: string,
    cutoffDate: Date
  ): Promise<AuditRecord[]> {
    const result = await this.pool.query(
      `SELECT * FROM audit_trail
       WHERE tenant_id = $1 AND timestamp < $2
       ORDER BY timestamp ASC`,
      [tenantId, cutoffDate]
    );

    return result.rows.map(this.mapRowToAuditRecord);
  }

  private async loadToBigQuery(
    tenantId: string,
    records: AuditRecord[]
  ): Promise<void> {
    const dataset = this.bigquery.dataset('audit_archive');
    const table = dataset.table('audit_trail');

    await table.insert(records.map(r => ({
      id: r.id,
      timestamp: r.timestamp.toISOString(),
      tenant_id: r.tenantId,
      user_id: r.userId,
      session_id: r.sessionId,
      action_type: r.actionType,
      resource_type: r.resourceType,
      resource_id: r.resourceId,
      old_value: JSON.stringify(r.oldValue),
      new_value: JSON.stringify(r.newValue),
      metadata: JSON.stringify(r.metadata),
      hash: r.hash,
      previous_hash: r.previousHash,
      created_at: r.createdAt.toISOString(),
      ip_address: r.ipAddress,
      user_agent: r.userAgent,
    })));
  }

  private async countBigQueryRecords(
    tenantId: string,
    startDate: Date,
    endDate: Date
  ): Promise<number> {
    const query = `
      SELECT COUNT(*) as count
      FROM \`audit_archive.audit_trail\`
      WHERE tenant_id = @tenantId
        AND timestamp >= @startDate
        AND timestamp <= @endDate
    `;

    const [rows] = await this.bigquery.query({
      query,
      params: {
        tenantId,
        startDate: startDate.toISOString(),
        endDate: endDate.toISOString(),
      },
    });

    return parseInt(rows[0].count, 10);
  }

  private async deleteHotRecords(
    tenantId: string,
    cutoffDate: Date
  ): Promise<void> {
    // Note: This is the ONLY permitted deletion (archival after verification)
    // Requires special superadmin permission with separate audit trail

    await this.pool.query(
      `DELETE FROM audit_trail
       WHERE tenant_id = $1 AND timestamp < $2`,
      [tenantId, cutoffDate]
    );
  }

  private async exportBigQueryToParquet(
    tenantId: string,
    cutoffDate: Date
  ): Promise<string> {
    const exportUri = `gs://audit-archive-cold/${tenantId}/${cutoffDate.getFullYear()}/audit_trail_*.parquet`;

    const query = `
      EXPORT DATA OPTIONS(
        uri='${exportUri}',
        format='PARQUET',
        compression='SNAPPY'
      ) AS
      SELECT * FROM \`audit_archive.audit_trail\`
      WHERE tenant_id = '${tenantId}' AND timestamp < '${cutoffDate.toISOString()}'
    `;

    await this.bigquery.query(query);

    return exportUri;
  }

  private async moveToArchiveClass(exportUri: string): Promise<void> {
    // Change storage class to Archive (cheapest, 365-day minimum)
    const bucket = this.storage.bucket('audit-archive-cold');

    // Extract glob pattern files
    const [files] = await bucket.getFiles({ prefix: exportUri });

    for (const file of files) {
      await file.setStorageClass('ARCHIVE');
    }
  }

  private async verifyColdStorage(exportUri: string): Promise<boolean> {
    // Verify files exist and checksums match
    const bucket = this.storage.bucket('audit-archive-cold');
    const [files] = await bucket.getFiles({ prefix: exportUri });

    return files.length > 0;
  }

  private async deleteBigQueryRecords(
    tenantId: string,
    cutoffDate: Date
  ): Promise<void> {
    const query = `
      DELETE FROM \`audit_archive.audit_trail\`
      WHERE tenant_id = '${tenantId}' AND timestamp < '${cutoffDate.toISOString()}'
    `;

    await this.bigquery.query(query);
  }

  private async queryHotStorage(
    tenantId: string,
    startDate: Date,
    endDate: Date
  ): Promise<AuditRecord[]> {
    const result = await this.pool.query(
      `SELECT * FROM audit_trail
       WHERE tenant_id = $1
         AND timestamp >= $2
         AND timestamp <= $3
       ORDER BY timestamp ASC`,
      [tenantId, startDate, endDate]
    );

    return result.rows.map(this.mapRowToAuditRecord);
  }

  private async queryWarmStorage(
    tenantId: string,
    startDate: Date,
    endDate: Date
  ): Promise<AuditRecord[]> {
    const query = `
      SELECT * FROM \`audit_archive.audit_trail\`
      WHERE tenant_id = @tenantId
        AND timestamp >= @startDate
        AND timestamp <= @endDate
      ORDER BY timestamp ASC
    `;

    const [rows] = await this.bigquery.query({
      query,
      params: {
        tenantId,
        startDate: startDate.toISOString(),
        endDate: endDate.toISOString(),
      },
    });

    return rows.map(this.mapBigQueryRowToAuditRecord);
  }

  private async queryColdStorage(
    tenantId: string,
    startDate: Date,
    endDate: Date
  ): Promise<AuditRecord[]> {
    // Load Parquet files from GCS Archive into temporary BigQuery table
    // Query and return results
    throw new Error('Cold storage query not implemented');
  }

  private mapRowToAuditRecord(row: any): AuditRecord {
    return {
      id: row.id,
      timestamp: new Date(row.timestamp),
      tenantId: row.tenant_id,
      userId: row.user_id,
      sessionId: row.session_id,
      actionType: row.action_type as AuditActionType,
      resourceType: row.resource_type as AuditResourceType,
      resourceId: row.resource_id,
      oldValue: row.old_value,
      newValue: row.new_value,
      metadata: row.metadata,
      hash: row.hash,
      previousHash: row.previous_hash,
      createdAt: new Date(row.created_at),
      ipAddress: row.ip_address,
      userAgent: row.user_agent,
    };
  }

  private mapBigQueryRowToAuditRecord(row: any): AuditRecord {
    return {
      id: row.id,
      timestamp: new Date(row.timestamp),
      tenantId: row.tenant_id,
      userId: row.user_id,
      sessionId: row.session_id,
      actionType: row.action_type as AuditActionType,
      resourceType: row.resource_type as AuditResourceType,
      resourceId: row.resource_id,
      oldValue: row.old_value ? JSON.parse(row.old_value) : undefined,
      newValue: row.new_value ? JSON.parse(row.new_value) : undefined,
      metadata: JSON.parse(row.metadata),
      hash: row.hash,
      previousHash: row.previous_hash,
      createdAt: new Date(row.created_at),
      ipAddress: row.ip_address,
      userAgent: row.user_agent,
    };
  }

  private async logArchivalEvent(event: any): Promise<void> {
    // Log to separate archival_events table for audit trail of the audit trail
    await this.pool.query(
      `INSERT INTO archival_events (tenant_id, event_type, event_data, created_at)
       VALUES ($1, $2, $3, NOW())`,
      [event.tenantId, 'archival', JSON.stringify(event)]
    );
  }
}

5.3 Automated Archival Cron Jobs​

// ============================================================================
// CRON SCHEDULES (Cloud Scheduler)
// ============================================================================

import { CloudSchedulerClient } from '@google-cloud/scheduler';

export class ArchivalScheduler {
  private scheduler: CloudSchedulerClient;

  constructor() {
    this.scheduler = new CloudSchedulerClient();
  }

  /**
   * Setup automated archival jobs
   */
  async setupArchivalJobs(): Promise<void> {
    // Daily hot-to-warm archival (02:00 UTC)
    await this.createJob({
      name: 'audit-archival-hot-to-warm',
      schedule: '0 2 * * *',  // Daily at 2 AM
      endpoint: '/api/internal/audit/archive/hot-to-warm',
      description: 'Archive audit records from PostgreSQL to BigQuery',
    });

    // Weekly warm-to-cold archival (Sunday 03:00 UTC)
    await this.createJob({
      name: 'audit-archival-warm-to-cold',
      schedule: '0 3 * * 0',  // Weekly on Sunday at 3 AM
      endpoint: '/api/internal/audit/archive/warm-to-cold',
      description: 'Archive audit records from BigQuery to GCS Archive',
    });

    // Daily hash chain verification (01:00 UTC)
    await this.createJob({
      name: 'audit-hash-chain-verification',
      schedule: '0 1 * * *',  // Daily at 1 AM
      endpoint: '/api/internal/audit/verify/hash-chain',
      description: 'Verify audit trail hash chain integrity',
    });

    // Daily Merkle root publication (04:00 UTC)
    await this.createJob({
      name: 'audit-merkle-root-publication',
      schedule: '0 4 * * *',  // Daily at 4 AM
      endpoint: '/api/internal/audit/publish/merkle-root',
      description: 'Publish daily Merkle root hash to external store',
    });
  }

  private async createJob(config: {
    name: string;
    schedule: string;
    endpoint: string;
    description: string;
  }): Promise<void> {
    // Implementation: Create Cloud Scheduler job with OIDC auth
  }
}

6. Performance Optimization​

6.1 Write Performance​

Target: <5ms p95 latency for audit record insertion

Optimizations:

  1. Partitioning: Monthly partitions reduce index size and improve write throughput
  2. Async Hash Computation: Trigger computes hash synchronously, but Merkle tree is async
  3. Connection Pooling: Reuse database connections (pgBouncer in transaction mode)
  4. Bulk Inserts: Batch audit records where possible (e.g., bulk document imports)
// ============================================================================
// BULK AUDIT INSERTION
// ============================================================================

export class AuditService {
  /**
   * Bulk insert audit records (e.g., during data migrations)
   * Uses COPY command for maximum throughput
   */
  async bulkInsert(records: Omit<AuditRecord, 'id' | 'hash' | 'previousHash'>[]): Promise<void> {
    const client = await this.pool.connect();

    try {
      await client.query('BEGIN');

      // Create temporary table
      await client.query(`
        CREATE TEMP TABLE temp_audit_records (
          timestamp TIMESTAMPTZ,
          tenant_id UUID,
          user_id UUID,
          session_id UUID,
          action_type VARCHAR(50),
          resource_type VARCHAR(100),
          resource_id UUID,
          old_value JSONB,
          new_value JSONB,
          metadata JSONB,
          created_at TIMESTAMPTZ,
          ip_address INET,
          user_agent TEXT
        )
      `);

      // Use COPY for bulk load
      const copyStream = client.query(
        copyFrom('COPY temp_audit_records FROM STDIN WITH CSV')
      );

      for (const record of records) {
        copyStream.write([
          record.timestamp.toISOString(),
          record.tenantId,
          record.userId ?? null,
          record.sessionId ?? null,
          record.actionType,
          record.resourceType,
          record.resourceId ?? null,
          JSON.stringify(record.oldValue ?? null),
          JSON.stringify(record.newValue ?? null),
          JSON.stringify(record.metadata),
          record.createdAt.toISOString(),
          record.ipAddress ?? null,
          record.userAgent ?? null,
        ].join(',') + '\n');
      }

      copyStream.end();

      // Insert from temp table (triggers will compute hashes)
      await client.query(`
        INSERT INTO audit_trail (
          timestamp, tenant_id, user_id, session_id, action_type,
          resource_type, resource_id, old_value, new_value, metadata,
          created_at, ip_address, user_agent
        )
        SELECT * FROM temp_audit_records
        ORDER BY timestamp ASC
      `);

      await client.query('COMMIT');
    } catch (error) {
      await client.query('ROLLBACK');
      throw error;
    } finally {
      client.release();
    }
  }
}

6.2 Query Performance​

Common Queries & Indexes:

-- Query 1: User activity audit (most common)
-- Uses: idx_audit_user
SELECT * FROM audit_trail
WHERE tenant_id = $1 AND user_id = $2
  AND timestamp >= $3 AND timestamp <= $4
ORDER BY timestamp DESC
LIMIT 100;

-- Query 2: Resource change history
-- Uses: idx_audit_resource
SELECT * FROM audit_trail
WHERE tenant_id = $1
  AND resource_type = $2
  AND resource_id = $3
ORDER BY timestamp DESC;

-- Query 3: Compliance report (all actions in date range)
-- Uses: idx_audit_tenant_timestamp
SELECT
  action_type,
  COUNT(*) as action_count,
  COUNT(DISTINCT user_id) as unique_users
FROM audit_trail
WHERE tenant_id = $1
  AND timestamp >= $2
  AND timestamp <= $3
GROUP BY action_type;

-- Query 4: High-risk action monitoring
-- Uses: idx_audit_action_type
SELECT * FROM audit_trail
WHERE tenant_id = $1
  AND action_type IN ('DELETE', 'APPROVE', 'SIGN')
  AND timestamp >= NOW() - INTERVAL '7 days'
ORDER BY timestamp DESC;

6.3 Performance Monitoring​

// ============================================================================
// PERFORMANCE METRICS COLLECTOR
// ============================================================================

import { Histogram } from 'prom-client';

export class AuditPerformanceMonitor {
  private writeLatencyHistogram: Histogram<string>;
  private queryLatencyHistogram: Histogram<string>;

  constructor() {
    this.writeLatencyHistogram = new Histogram({
      name: 'audit_write_latency_ms',
      help: 'Audit record write latency in milliseconds',
      labelNames: ['tenant_id', 'action_type'],
      buckets: [1, 2, 5, 10, 20, 50, 100, 200, 500],
    });

    this.queryLatencyHistogram = new Histogram({
      name: 'audit_query_latency_ms',
      help: 'Audit query latency in milliseconds',
      labelNames: ['tenant_id', 'query_type'],
      buckets: [10, 25, 50, 100, 250, 500, 1000, 2500, 5000],
    });
  }

  /**
   * Track write latency (should be <5ms p95)
   */
  async trackWrite<T>(
    tenantId: string,
    actionType: string,
    operation: () => Promise<T>
  ): Promise<T> {
    const start = Date.now();

    try {
      const result = await operation();
      const latency = Date.now() - start;

      this.writeLatencyHistogram.observe(
        { tenant_id: tenantId, action_type: actionType },
        latency
      );

      // Alert if >10ms (degraded performance)
      if (latency > 10) {
        console.warn(
          `Audit write latency degraded: ${latency}ms for ${actionType} (tenant ${tenantId})`
        );
      }

      return result;
    } catch (error) {
      const latency = Date.now() - start;
      this.writeLatencyHistogram.observe(
        { tenant_id: tenantId, action_type: actionType },
        latency
      );
      throw error;
    }
  }

  /**
   * Track query latency
   */
  async trackQuery<T>(
    tenantId: string,
    queryType: string,
    operation: () => Promise<T>
  ): Promise<T> {
    const start = Date.now();

    try {
      const result = await operation();
      const latency = Date.now() - start;

      this.queryLatencyHistogram.observe(
        { tenant_id: tenantId, query_type: queryType },
        latency
      );

      return result;
    } catch (error) {
      const latency = Date.now() - start;
      this.queryLatencyHistogram.observe(
        { tenant_id: tenantId, query_type: queryType },
        latency
      );
      throw error;
    }
  }
}

7. API Integration​

7.1 AuditService Core API​

// ============================================================================
// AUDIT SERVICE - PRIMARY API
// ============================================================================

import { Pool } from 'pg';
import { v4 as uuidv4 } from 'uuid';

export class AuditService {
  constructor(
    private pool: Pool,
    private tenantContext: TenantContextManager,
    private performanceMonitor: AuditPerformanceMonitor
  ) {}

  /**
   * Record an audit event
   *
   * Usage:
   *   await auditService.record({
   *     tenantId: req.tenantId,
   *     userId: req.user.id,
   *     sessionId: req.sessionId,
   *     actionType: AuditActionType.UPDATE,
   *     resourceType: AuditResourceType.SOP,
   *     resourceId: sop.id,
   *     oldValue: { status: 'draft' },
   *     newValue: { status: 'approved' },
   *     metadata: {
   *       description: 'SOP approved by quality manager',
   *       approvalChain: ['user-1', 'user-2'],
   *     },
   *   });
   */
  async record(params: {
    tenantId: string;
    userId?: string;
    sessionId?: string;
    actionType: AuditActionType;
    resourceType: AuditResourceType;
    resourceId?: string;
    oldValue?: Record<string, any>;
    newValue?: Record<string, any>;
    metadata?: AuditMetadata;
    ipAddress?: string;
    userAgent?: string;
  }): Promise<AuditRecord> {
    return await this.performanceMonitor.trackWrite(
      params.tenantId,
      params.actionType,
      async () => {
        return await this.tenantContext.withTenantContext(
          params.tenantId,
          async (client) => {
            const result = await client.query(
              `INSERT INTO audit_trail (
                timestamp,
                tenant_id,
                user_id,
                session_id,
                action_type,
                resource_type,
                resource_id,
                old_value,
                new_value,
                metadata,
                ip_address,
                user_agent
              ) VALUES (
                NOW(),
                $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11
              ) RETURNING *`,
              [
                params.tenantId,
                params.userId ?? null,
                params.sessionId ?? uuidv4(),
                params.actionType,
                params.resourceType,
                params.resourceId ?? null,
                params.oldValue ? JSON.stringify(params.oldValue) : null,
                params.newValue ? JSON.stringify(params.newValue) : null,
                JSON.stringify(params.metadata ?? {}),
                params.ipAddress ?? null,
                params.userAgent ?? null,
              ]
            );

            return this.mapRowToAuditRecord(result.rows[0]);
          }
        );
      }
    );
  }

  /**
   * Query audit records with filters
   */
  async query(filters: AuditQueryFilters): Promise<AuditQueryResult> {
    return await this.performanceMonitor.trackQuery(
      filters.tenantId,
      'filtered_query',
      async () => {
        return await this.tenantContext.withTenantContext(
          filters.tenantId,
          async (client) => {
            // Build dynamic query
            const conditions: string[] = ['tenant_id = $1'];
            const values: any[] = [filters.tenantId];
            let paramIndex = 2;

            if (filters.startDate) {
              conditions.push(`timestamp >= $${paramIndex++}`);
              values.push(filters.startDate);
            }

            if (filters.endDate) {
              conditions.push(`timestamp <= $${paramIndex++}`);
              values.push(filters.endDate);
            }

            if (filters.userId) {
              conditions.push(`user_id = $${paramIndex++}`);
              values.push(filters.userId);
            }

            if (filters.resourceId) {
              conditions.push(`resource_id = $${paramIndex++}`);
              values.push(filters.resourceId);
            }

            if (filters.actionTypes && filters.actionTypes.length > 0) {
              conditions.push(`action_type = ANY($${paramIndex++})`);
              values.push(filters.actionTypes);
            }

            if (filters.resourceTypes && filters.resourceTypes.length > 0) {
              conditions.push(`resource_type = ANY($${paramIndex++})`);
              values.push(filters.resourceTypes);
            }

            const whereClause = conditions.join(' AND ');
            const sortOrder = filters.sortOrder ?? 'DESC';
            const limit = filters.limit ?? 100;
            const offset = filters.offset ?? 0;

            // Count total matching records
            const countResult = await client.query(
              `SELECT COUNT(*) as total FROM audit_trail WHERE ${whereClause}`,
              values
            );

            const totalCount = parseInt(countResult.rows[0].total, 10);

            // Fetch paginated results
            const result = await client.query(
              `SELECT * FROM audit_trail
               WHERE ${whereClause}
               ORDER BY timestamp ${sortOrder}
               LIMIT $${paramIndex} OFFSET $${paramIndex + 1}`,
              [...values, limit, offset]
            );

            const records = result.rows.map(this.mapRowToAuditRecord);

            return {
              records,
              totalCount,
              hasMore: offset + records.length < totalCount,
              nextOffset: offset + records.length,
            };
          }
        );
      }
    );
  }

  /**
   * Get resource change history
   */
  async getResourceHistory(
    tenantId: string,
    resourceType: AuditResourceType,
    resourceId: string
  ): Promise<AuditRecord[]> {
    return await this.performanceMonitor.trackQuery(
      tenantId,
      'resource_history',
      async () => {
        return await this.tenantContext.withTenantContext(
          tenantId,
          async (client) => {
            const result = await client.query(
              `SELECT * FROM audit_trail
               WHERE tenant_id = $1
                 AND resource_type = $2
                 AND resource_id = $3
               ORDER BY timestamp ASC`,
              [tenantId, resourceType, resourceId]
            );

            return result.rows.map(this.mapRowToAuditRecord);
          }
        );
      }
    );
  }

  private mapRowToAuditRecord(row: any): AuditRecord {
    return {
      id: row.id,
      timestamp: new Date(row.timestamp),
      tenantId: row.tenant_id,
      userId: row.user_id,
      sessionId: row.session_id,
      actionType: row.action_type as AuditActionType,
      resourceType: row.resource_type as AuditResourceType,
      resourceId: row.resource_id,
      oldValue: row.old_value,
      newValue: row.new_value,
      metadata: row.metadata,
      hash: row.hash,
      previousHash: row.previous_hash,
      createdAt: new Date(row.created_at),
      ipAddress: row.ip_address,
      userAgent: row.user_agent,
    };
  }
}

7.2 Express Middleware for Automatic HTTP Auditing​

// ============================================================================
// AUDIT MIDDLEWARE - AUTOMATIC HTTP REQUEST AUDITING
// ============================================================================

import { Request, Response, NextFunction } from 'express';

export interface AuditableRequest extends Request {
  tenantId?: string;
  userId?: string;
  sessionId?: string;
}

export class AuditMiddleware {
  constructor(private auditService: AuditService) {}

  /**
   * Middleware to automatically audit all HTTP requests
   *
   * Usage:
   *   app.use(auditMiddleware.captureRequest());
   */
  captureRequest() {
    return async (req: AuditableRequest, res: Response, next: NextFunction) => {
      // Skip non-auditable requests
      if (!req.tenantId || this.shouldSkipAudit(req)) {
        return next();
      }

      // Capture request start time
      const startTime = Date.now();

      // Capture original response methods
      const originalJson = res.json.bind(res);
      const originalSend = res.send.bind(res);

      let responseBody: any;

      // Intercept response
      res.json = (body: any) => {
        responseBody = body;
        return originalJson(body);
      };

      res.send = (body: any) => {
        responseBody = body;
        return originalSend(body);
      };

      // Capture response completion
      res.on('finish', async () => {
        try {
          const duration = Date.now() - startTime;

          await this.auditService.record({
            tenantId: req.tenantId!,
            userId: req.userId,
            sessionId: req.sessionId,
            actionType: this.mapMethodToAction(req.method),
            resourceType: this.extractResourceType(req.path),
            resourceId: this.extractResourceId(req.path),
            oldValue: req.method === 'PUT' || req.method === 'PATCH'
              ? this.extractOldValue(req, responseBody)
              : undefined,
            newValue: req.method !== 'GET' && req.method !== 'DELETE'
              ? this.extractNewValue(req, responseBody)
              : undefined,
            metadata: {
              method: req.method,
              endpoint: req.path,
              statusCode: res.statusCode,
              durationMs: duration,
              query: req.query,
              userAgent: req.get('user-agent'),
            },
            ipAddress: req.ip,
            userAgent: req.get('user-agent'),
          });
        } catch (error) {
          console.error('Audit middleware error:', error);
          // Don't fail the request on audit errors
        }
      });

      next();
    };
  }

  private shouldSkipAudit(req: Request): boolean {
    // Skip health checks, metrics, static assets
    const skipPaths = ['/health', '/metrics', '/static', '/favicon.ico'];
    return skipPaths.some(path => req.path.startsWith(path));
  }

  private mapMethodToAction(method: string): AuditActionType {
    switch (method.toUpperCase()) {
      case 'POST': return AuditActionType.CREATE;
      case 'PUT':
      case 'PATCH': return AuditActionType.UPDATE;
      case 'DELETE': return AuditActionType.DELETE;
      case 'GET': return AuditActionType.READ;
      default: return AuditActionType.EXECUTE;
    }
  }

  private extractResourceType(path: string): AuditResourceType {
    // Extract from path: /api/documents/123 -> document
    const match = path.match(/\/api\/([^\/]+)/);
    if (!match) return AuditResourceType.DOCUMENT;

    const segment = match[1];

    // Map plural to singular resource types
    const typeMap: Record<string, AuditResourceType> = {
      documents: AuditResourceType.DOCUMENT,
      sops: AuditResourceType.SOP,
      capas: AuditResourceType.CAPA,
      deviations: AuditResourceType.DEVIATION,
      'change-controls': AuditResourceType.CHANGE_CONTROL,
      'training-records': AuditResourceType.TRAINING_RECORD,
      'validation-protocols': AuditResourceType.VALIDATION_PROTOCOL,
    };

    return typeMap[segment] ?? AuditResourceType.DOCUMENT;
  }

  private extractResourceId(path: string): string | undefined {
    // Extract UUID from path: /api/documents/550e8400-e29b-41d4-a716-446655440000
    const match = path.match(/([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})/i);
    return match ? match[1] : undefined;
  }

  private extractOldValue(req: any, responseBody: any): Record<string, any> | undefined {
    // For updates, capture "before" state from response (if included)
    return responseBody?.previousValue ?? undefined;
  }

  private extractNewValue(req: any, responseBody: any): Record<string, any> | undefined {
    // Capture request body for creates/updates
    return req.body ?? responseBody;
  }
}

7.3 Database Trigger-Based Auditing​

-- ============================================================================
-- AUTOMATIC AUDIT FOR CRITICAL TABLES
-- ============================================================================

-- Example: Auto-audit SOP changes at database level
CREATE OR REPLACE FUNCTION audit_sop_changes()
RETURNS TRIGGER AS $$
BEGIN
    IF TG_OP = 'INSERT' THEN
        INSERT INTO audit_trail (
            tenant_id,
            user_id,
            action_type,
            resource_type,
            resource_id,
            new_value,
            metadata
        ) VALUES (
            NEW.tenant_id,
            current_setting('app.current_user_id', true)::uuid,
            'CREATE',
            'sop',
            NEW.id,
            to_jsonb(NEW),
            jsonb_build_object('trigger_source', 'database')
        );
    ELSIF TG_OP = 'UPDATE' THEN
        INSERT INTO audit_trail (
            tenant_id,
            user_id,
            action_type,
            resource_type,
            resource_id,
            old_value,
            new_value,
            metadata
        ) VALUES (
            NEW.tenant_id,
            current_setting('app.current_user_id', true)::uuid,
            'UPDATE',
            'sop',
            NEW.id,
            to_jsonb(OLD),
            to_jsonb(NEW),
            jsonb_build_object('trigger_source', 'database')
        );
    ELSIF TG_OP = 'DELETE' THEN
        INSERT INTO audit_trail (
            tenant_id,
            user_id,
            action_type,
            resource_type,
            resource_id,
            old_value,
            metadata
        ) VALUES (
            OLD.tenant_id,
            current_setting('app.current_user_id', true)::uuid,
            'DELETE',
            'sop',
            OLD.id,
            to_jsonb(OLD),
            jsonb_build_object('trigger_source', 'database')
        );
    END IF;

    RETURN NULL;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER trg_audit_sop_changes
    AFTER INSERT OR UPDATE OR DELETE ON sops
    FOR EACH ROW
    EXECUTE FUNCTION audit_sop_changes();

8. Verification & Monitoring​

8.1 Daily Hash Chain Verification​

// ============================================================================
// HASH CHAIN VERIFICATION SERVICE
// ============================================================================

export class HashChainVerificationService {
  constructor(
    private auditService: AuditService,
    private externalHashStore: ExternalHashStore,
    private alertService: AlertService
  ) {}

  /**
   * Run daily hash chain verification for all tenants
   * Scheduled via Cloud Scheduler at 01:00 UTC
   */
  async runDailyVerification(): Promise<void> {
    const tenants = await this.getAllTenants();

    const results = await Promise.all(
      tenants.map(tenant => this.verifyTenantHashChain(tenant.id))
    );

    // Alert on failures
    const failures = results.filter(r => r.failedRecords > 0);
    if (failures.length > 0) {
      await this.alertService.sendCriticalAlert({
        title: 'Audit Trail Integrity Failure',
        message: `Hash chain verification failed for ${failures.length} tenant(s)`,
        details: failures,
        severity: 'CRITICAL',
      });
    }

    // Log verification results
    await this.logVerificationResults(results);
  }

  /**
   * Verify hash chain for a single tenant
   */
  async verifyTenantHashChain(tenantId: string): Promise<HashChainVerificationResult> {
    // Fetch last 24 hours of records
    const yesterday = new Date();
    yesterday.setDate(yesterday.getDate() - 1);

    const records = await this.auditService.query({
      tenantId,
      startDate: yesterday,
      endDate: new Date(),
      limit: 1000000, // No limit
    });

    // Verify hash chain
    const verification = HashChainBuilder.verifyChain(records.records);

    // If successful, publish Merkle root to external store
    if (verification.failedRecords === 0) {
      await this.externalHashStore.publishDailyRootHash(
        tenantId,
        yesterday,
        verification.merkleRootHash,
        verification.totalRecords
      );
    }

    return verification;
  }

  private async getAllTenants(): Promise<Array<{ id: string }>> {
    // Fetch all active tenants
    return [];
  }

  private async logVerificationResults(
    results: HashChainVerificationResult[]
  ): Promise<void> {
    // Store in verification_results table
  }
}

8.2 Monitoring Alerts​

// ============================================================================
// AUDIT MONITORING ALERTS
// ============================================================================

export interface AlertConfig {
  name: string;
  condition: (metrics: AuditMetrics) => boolean;
  severity: 'INFO' | 'WARNING' | 'CRITICAL';
  message: string;
}

export class AuditMonitoringService {
  private alerts: AlertConfig[] = [
    {
      name: 'high_write_latency',
      condition: (m) => m.writeLatencyP95 > 10,
      severity: 'WARNING',
      message: 'Audit write latency degraded (p95 > 10ms)',
    },
    {
      name: 'hash_chain_broken',
      condition: (m) => m.hashChainFailures > 0,
      severity: 'CRITICAL',
      message: 'Audit trail integrity compromised - hash chain broken',
    },
    {
      name: 'archival_failure',
      condition: (m) => m.archivalFailures > 0,
      severity: 'CRITICAL',
      message: 'Audit archival pipeline failed',
    },
    {
      name: 'high_audit_volume',
      condition: (m) => m.recordsPerSecond > 1000,
      severity: 'WARNING',
      message: 'Unusually high audit record volume',
    },
    {
      name: 'rls_bypass_attempt',
      condition: (m) => m.rlsBypassAttempts > 0,
      severity: 'CRITICAL',
      message: 'Cross-tenant audit access attempted',
    },
  ];

  /**
   * Run monitoring checks every 5 minutes
   */
  async runMonitoring(): Promise<void> {
    const metrics = await this.collectMetrics();

    for (const alert of this.alerts) {
      if (alert.condition(metrics)) {
        await this.sendAlert({
          name: alert.name,
          severity: alert.severity,
          message: alert.message,
          metrics,
        });
      }
    }
  }

  private async collectMetrics(): Promise<AuditMetrics> {
    // Collect from Prometheus/Cloud Monitoring
    return {
      writeLatencyP95: 0,
      hashChainFailures: 0,
      archivalFailures: 0,
      recordsPerSecond: 0,
      rlsBypassAttempts: 0,
    };
  }

  private async sendAlert(alert: any): Promise<void> {
    // Send to PagerDuty/Slack/Email
  }
}

interface AuditMetrics {
  writeLatencyP95: number;
  hashChainFailures: number;
  archivalFailures: number;
  recordsPerSecond: number;
  rlsBypassAttempts: number;
}

9. Compliance Controls​

9.1 FDA 21 CFR Part 11 Compliance​

RequirementSectionImplementationEvidence
Audit trails for all record changes§11.10(e)Hash-chained append-only audit_trail tableDatabase schema + triggers
Secure, computer-generated, time-stamped§11.10(e)(1)PostgreSQL NOW(), timestamptz, NTP syncServer NTP config
Operator identification§11.10(e)(2)user_id, session_id, ip_addressaudit_trail.user_id
Action performed§11.10(e)(3)action_type enum, resource_typeaudit_trail.action_type
Date and time stamp§11.10(e)(4)timestamp (UTC)audit_trail.timestamp
Previous values§11.10(e)(5)old_value JSONBaudit_trail.old_value
Not obscure audit trail§11.10(e)(6)Immutable, no overwritesDatabase triggers
7-year retention§11.10(g)Hot/warm/cold archivalAuditArchivalService

Validation Evidence:

  • Database schema DDL with immutability triggers
  • Hash chain verification logs (daily)
  • Archival pipeline execution logs
  • Performance benchmarks (<5ms write latency)
  • RLS policy test results

9.2 HIPAA Security Rule Compliance​

StandardSectionImplementationEvidence
Audit Controls§164.312(b)RLS-enforced multi-tenant audit loggingRLS policies + tests
Integrity§164.312(c)(1)Cryptographic hash chainHash verification logs
Person or Entity Authentication§164.312(d)user_id, session_id, MFAAuthentication logs
Access Audit§164.308(a)(1)(ii)(D)Query API for audit reportsAPI endpoints
6-year retention§164.316(b)(2)(i)Warm/cold storage tiersArchival logs

BAA Requirements:

  • Audit access to PHI recorded with tenant_id isolation
  • Break-glass superadmin access separately audited
  • Encryption at rest (database encryption) and in transit (TLS)

9.3 SOC 2 Type II Compliance​

CriterionControlImplementationTest Procedure
CC7.2System monitoring for anomaliesDaily hash chain verificationVerification logs
CC7.3Monitoring activities evaluationAutomated alert thresholdsAlert configuration
CC7.4Monitoring results remediationPagerDuty alerts + runbooksIncident response logs
CC6.1Logical access restrictionsRLS policies + tenant isolationRLS bypass tests
CC6.6Audit loggingComprehensive audit trailAudit record sampling

Type II Evidence (6-month observation period):

  • Monthly hash chain verification reports
  • Archival pipeline execution logs
  • Performance SLA compliance (p95 <5ms)
  • Incident response for integrity failures (0 expected)

10. Implementation Guide​

10.1 Deployment Checklist​

Phase 1: Database Setup​

  • Create audit_trail table with partitioning
  • Create immutability triggers (prevent_audit_update, prevent_audit_delete)
  • Create hash computation trigger (compute_audit_hash)
  • Create RLS policies (audit_tenant_isolation)
  • Create indexes (tenant_timestamp, resource, user, hash_chain)
  • Test write performance (<5ms p95)
  • Test RLS isolation (cross-tenant access blocked)

Phase 2: Application Integration​

  • Implement AuditService.record() API
  • Implement AuditService.query() API
  • Add AuditMiddleware to Express app
  • Add database triggers for critical tables (SOPs, CAPAs, etc.)
  • Test hash chain computation (manual verification)
  • Test bulk insert performance (1000 records <1s)

Phase 3: Hash Chain Verification​

  • Implement HashChainBuilder.verifyChain()
  • Implement HashChainBuilder.computeMerkleRoot()
  • Implement ExternalHashStore (Cloud KMS signing)
  • Setup Cloud Scheduler daily verification job
  • Setup Cloud Scheduler daily Merkle root publication
  • Test end-to-end verification (tamper detection)

Phase 4: Archival Pipeline​

  • Implement AuditArchivalService.archiveToWarmStorage()
  • Implement AuditArchivalService.archiveToColdStorage()
  • Create BigQuery audit_archive dataset + table
  • Create GCS audit-archive-cold bucket (Archive class)
  • Setup Cloud Scheduler archival jobs
  • Test hot-to-warm archival (integrity verification)
  • Test warm-to-cold archival (Parquet export)
  • Test restore from archive (cross-tier query)

Phase 5: Monitoring & Alerting​

  • Implement AuditPerformanceMonitor (Prometheus metrics)
  • Implement AuditMonitoringService (alert thresholds)
  • Setup PagerDuty integration for critical alerts
  • Create Grafana dashboards (write latency, hash chain status)
  • Test alert conditions (simulate hash chain break)
  • Document runbooks for integrity failures

Phase 6: Compliance Validation​

  • Generate FDA 21 CFR Part 11 validation protocol
  • Generate HIPAA audit report sample
  • Generate SOC 2 Type II evidence package
  • Conduct penetration testing (RLS bypass attempts)
  • Conduct performance testing (load test 1M records)
  • Document audit trail verification procedures

10.2 Testing Strategy​

Unit Tests​

describe('AuditService', () => {
  it('should insert audit record with correct hash', async () => {
    const record = await auditService.record({
      tenantId: 'tenant-1',
      userId: 'user-1',
      actionType: AuditActionType.CREATE,
      resourceType: AuditResourceType.DOCUMENT,
      newValue: { title: 'Test Doc' },
    });

    expect(record.hash).toMatch(/^[0-9a-f]{64}$/);
    expect(record.previousHash).toBeDefined();
  });

  it('should enforce RLS tenant isolation', async () => {
    await expect(
      auditService.query({
        tenantId: 'tenant-2', // Different tenant
        resourceId: 'tenant-1-resource',
      })
    ).resolves.toHaveProperty('records', []);
  });
});

describe('HashChainBuilder', () => {
  it('should detect tampered record', () => {
    const records = [
      { hash: 'aaa...', previousHash: '000...', timestamp: new Date('2026-01-01') },
      { hash: 'bbb...', previousHash: 'aaa...', timestamp: new Date('2026-01-02') },
      { hash: 'ccc...', previousHash: 'xxx...', timestamp: new Date('2026-01-03') }, // Tampered
    ];

    const result = HashChainBuilder.verifyChain(records);
    expect(result.failedRecords).toBe(1);
    expect(result.failures[0].reason).toContain('Previous hash mismatch');
  });
});

Integration Tests​

describe('Audit Archival Pipeline', () => {
  it('should archive records to BigQuery', async () => {
    // Insert 100 records dated 91 days ago
    const records = await seedOldRecords(tenantId, 100, 91);

    // Run archival
    await archivalService.archiveToWarmStorage(tenantId);

    // Verify BigQuery load
    const count = await countBigQueryRecords(tenantId, startDate, endDate);
    expect(count).toBe(100);

    // Verify PostgreSQL deletion
    const pgCount = await countPostgresRecords(tenantId, startDate, endDate);
    expect(pgCount).toBe(0);
  });
});

Performance Tests​

describe('Audit Performance', () => {
  it('should maintain <5ms write latency at p95', async () => {
    const latencies: number[] = [];

    for (let i = 0; i < 1000; i++) {
      const start = Date.now();
      await auditService.record({
        tenantId: 'tenant-1',
        actionType: AuditActionType.CREATE,
        resourceType: AuditResourceType.DOCUMENT,
      });
      latencies.push(Date.now() - start);
    }

    const p95 = calculatePercentile(latencies, 0.95);
    expect(p95).toBeLessThan(5);
  });
});

Appendix A: Database Migration Scripts​

-- ============================================================================
-- MIGRATION 001: CREATE AUDIT_TRAIL TABLE
-- ============================================================================

BEGIN;

-- Create enum types
CREATE TYPE audit_action_type AS ENUM (
    'CREATE', 'UPDATE', 'DELETE', 'READ',
    'EXECUTE', 'APPROVE', 'REJECT', 'SIGN', 'EXPORT'
);

-- Create partitioned table
CREATE TABLE audit_trail (
    id                  BIGSERIAL PRIMARY KEY,
    timestamp           TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    tenant_id           UUID NOT NULL REFERENCES tenants(id),
    user_id             UUID REFERENCES users(id),
    session_id          UUID,
    action_type         audit_action_type NOT NULL,
    resource_type       VARCHAR(100) NOT NULL,
    resource_id         UUID,
    old_value           JSONB,
    new_value           JSONB,
    metadata            JSONB NOT NULL DEFAULT '{}'::jsonb,
    hash                CHAR(64) NOT NULL,
    previous_hash       CHAR(64),
    created_at          TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    ip_address          INET,
    user_agent          TEXT
) PARTITION BY RANGE (timestamp);

-- Create initial partitions (2 years)
DO $$
DECLARE
    start_date DATE := '2026-01-01';
    end_date DATE;
    partition_name TEXT;
BEGIN
    FOR i IN 0..23 LOOP
        end_date := start_date + INTERVAL '1 month';
        partition_name := 'audit_trail_' || TO_CHAR(start_date, 'YYYY_MM');

        EXECUTE format(
            'CREATE TABLE %I PARTITION OF audit_trail FOR VALUES FROM (%L) TO (%L)',
            partition_name,
            start_date,
            end_date
        );

        start_date := end_date;
    END LOOP;
END $$;

-- Create indexes
CREATE INDEX idx_audit_tenant_timestamp ON audit_trail (tenant_id, timestamp DESC);
CREATE INDEX idx_audit_resource ON audit_trail (tenant_id, resource_type, resource_id, timestamp DESC);
CREATE INDEX idx_audit_user ON audit_trail (tenant_id, user_id, timestamp DESC);
CREATE INDEX idx_audit_hash_chain ON audit_trail (tenant_id, timestamp ASC) INCLUDE (hash, previous_hash);
CREATE INDEX idx_audit_action_type ON audit_trail (tenant_id, action_type, timestamp DESC);
CREATE INDEX idx_audit_metadata_gin ON audit_trail USING GIN (metadata jsonb_path_ops);

-- Enable RLS
ALTER TABLE audit_trail ENABLE ROW LEVEL SECURITY;

CREATE POLICY audit_tenant_isolation ON audit_trail
    FOR SELECT
    USING (tenant_id = current_setting('app.current_tenant_id', true)::uuid);

-- Create immutability triggers
CREATE OR REPLACE FUNCTION prevent_audit_update()
RETURNS TRIGGER AS $func$
BEGIN
    RAISE EXCEPTION 'Audit records are immutable - UPDATE operations are not permitted';
END;
$func$ LANGUAGE plpgsql;

CREATE TRIGGER trg_prevent_audit_update
    BEFORE UPDATE ON audit_trail
    FOR EACH ROW
    EXECUTE FUNCTION prevent_audit_update();

CREATE OR REPLACE FUNCTION prevent_audit_delete()
RETURNS TRIGGER AS $func$
BEGIN
    RAISE EXCEPTION 'Audit records are immutable - DELETE operations are not permitted';
END;
$func$ LANGUAGE plpgsql;

CREATE TRIGGER trg_prevent_audit_delete
    BEFORE DELETE ON audit_trail
    FOR EACH ROW
    EXECUTE FUNCTION prevent_audit_delete();

-- Create hash computation trigger
CREATE OR REPLACE FUNCTION compute_audit_hash()
RETURNS TRIGGER AS $func$
DECLARE
    v_previous_hash CHAR(64);
    v_hash_input TEXT;
BEGIN
    SELECT hash INTO v_previous_hash
    FROM audit_trail
    WHERE tenant_id = NEW.tenant_id
    ORDER BY timestamp DESC, id DESC
    LIMIT 1;

    v_previous_hash := COALESCE(v_previous_hash, REPEAT('0', 64));

    v_hash_input := v_previous_hash ||
                    EXTRACT(EPOCH FROM NEW.timestamp)::TEXT ||
                    NEW.tenant_id::TEXT ||
                    COALESCE(NEW.user_id::TEXT, '') ||
                    NEW.action_type::TEXT ||
                    NEW.resource_type ||
                    COALESCE(NEW.resource_id::TEXT, '') ||
                    COALESCE(NEW.old_value::TEXT, '') ||
                    COALESCE(NEW.new_value::TEXT, '');

    NEW.hash := encode(digest(v_hash_input, 'sha256'), 'hex');
    NEW.previous_hash := v_previous_hash;

    RETURN NEW;
END;
$func$ LANGUAGE plpgsql;

CREATE TRIGGER trg_compute_audit_hash
    BEFORE INSERT ON audit_trail
    FOR EACH ROW
    EXECUTE FUNCTION compute_audit_hash();

COMMIT;

Appendix B: Regulatory Cross-Reference Matrix​

Control IDFDA 21 CFR Part 11HIPAASOC 2Implementation Component
D.5.1.1§11.10(e)§164.312(b)CC6.6audit_trail table schema
D.5.1.2§11.10(e)(6)§164.312(c)(1)CC7.2Immutability triggers
D.5.1.3§11.10(e)§164.312(b)CC7.2Hash chain computation
D.5.1.4-§164.312(a)(2)(i)CC6.1RLS tenant isolation
D.5.1.5§11.10(g)§164.316(b)(2)(i)CC6.6Archival pipeline
D.5.1.6--CC7.3Daily hash verification
D.5.1.7--CC7.4Monitoring alerts

Document Control​

VersionDateAuthorChanges
1.0.02026-02-16Claude (Sonnet 4.5)Initial comprehensive design

Review Schedule: Quarterly (Feb, May, Aug, Nov)

Next Review: 2026-05-16

Approval:

  • Chief Information Security Officer (CISO)
  • VP of Engineering
  • Director of Quality Assurance
  • Regulatory Affairs Manager

END OF DOCUMENT