Local Audit Logging (Core API Synchronous Write)

Overview

the Core API's audit middleware writes every mutation request (POST, PUT, PATCH, DELETE) synchronously to the local audit_log table. This is the HIPAA safety net — zero audit event loss is guaranteed, even if the Telemetry service is unavailable.

Why Synchronous?

HIPAA requirement: Audit controls (164.312(b)) mandate recording all activity in systems containing PHI. If the write is asynchronous and the application crashes before the event is persisted, we lose an audit entry — a compliance violation.

Performance trade-off: Synchronous writes add ~1ms latency to mutation requests. This is acceptable because:

Single INSERT statement (no joins or complex queries)
No RLS policy evaluation (audit_log has a simple CHECK (TRUE) policy)
Modern PostgreSQL INSERT is fast (~1ms for a single row)
Write uses connection pool (not the RLS-scoped connection), avoiding contention

Failed request logging: The audit write happens after the handler executes, so we log the response status code. Failed requests (401, 403, 500) are also audited — this is critical for security monitoring.

Architecture

HTTP Request (POST/PUT/PATCH/DELETE)
    ↓
Audit Middleware wraps handler
    ↓
Handler executes (business logic)
    ↓
Response captured (status code)
    ↓
Synchronous INSERT into audit_log (~1ms)
    ├─ Blocking operation
    ├─ Transaction guarantees durability
    └─ ERROR level log if this fails
    ↓
Async forward to Telemetry (non-blocking)
    ↓
Response sent to client

Implementation

Audit Middleware

// internal/core/middleware/audit.go

package middleware

import (
    "context"
    "net/http"
    "time"

    "github.com/jackc/pgx/v5/pgxpool"
    "github.com/restartix/internal/pkg/auditevt"
    "golang.org/x/exp/slog"
)

func AuditLog(db *pgxpool.Pool, forwarder *auditevt.Forwarder) func(http.Handler) http.Handler {
    return func(next http.Handler) http.Handler {
        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            // Skip GET requests (reads are not audited)
            if r.Method == http.MethodGet {
                next.ServeHTTP(w, r)
                return
            }

            user := UserFromContext(r.Context())
            if user == nil {
                // Unauthenticated requests are not audited (they'll fail at auth middleware)
                next.ServeHTTP(w, r)
                return
            }

            // Capture response using a response recorder
            recorder := &responseRecorder{
                ResponseWriter: w,
                statusCode:     http.StatusOK, // Default if WriteHeader is not called
            }

            // Execute handler
            next.ServeHTTP(recorder, r)

            // Build audit event
            event := auditevt.Event{
                Action:         mapMethodToAction(r.Method),
                EntityType:     extractEntityType(r.URL.Path),
                EntityID:       extractEntityID(r.URL.Path),
                UserID:         user.ID,
                OrganizationID: user.CurrentOrganizationID,
                IPAddress:      getClientIP(r),
                UserAgent:      r.UserAgent(),
                RequestPath:    r.Method + " " + r.URL.Path,
                StatusCode:     recorder.statusCode,
                Timestamp:      time.Now().UTC(),
            }

            // 1. SYNCHRONOUS: Write to local audit_log (HIPAA guarantee)
            if err := writeLocalAuditLog(r.Context(), db, event); err != nil {
                slog.Error("audit: local write failed", "error", err, "action", event.Action, "entity", event.EntityType)
                // This is a serious error — local audit must succeed.
                // Don't swallow it: log at ERROR level for alerting.
                // We DO NOT block the response — the mutation already happened.
                // But we want to know immediately if audit writes start failing.
            }

            // 2. ASYNC: Forward to Telemetry for enrichment + ClickHouse
            // Non-blocking — event is already safe in local audit_log
            forwarder.Forward(event)
        })
    }
}

// responseRecorder captures the status code from http.ResponseWriter
type responseRecorder struct {
    http.ResponseWriter
    statusCode int
}

func (r *responseRecorder) WriteHeader(statusCode int) {
    r.statusCode = statusCode
    r.ResponseWriter.WriteHeader(statusCode)
}

// writeLocalAuditLog inserts directly into the Core API's audit_log table.
// Uses the pool (not the RLS connection) because audit_log has no RLS.
func writeLocalAuditLog(ctx context.Context, db *pgxpool.Pool, event auditevt.Event) error {
    _, err := db.Exec(ctx, `
        INSERT INTO audit_log (
            organization_id, user_id, action, entity_type, entity_id,
            ip_address, user_agent, request_path, status_code, created_at
        ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
    `,
        event.OrganizationID, event.UserID, event.Action,
        event.EntityType, event.EntityID,
        event.IPAddress, event.UserAgent, event.RequestPath,
        event.StatusCode, event.Timestamp,
    )
    return err
}

// mapMethodToAction maps HTTP methods to CRUD actions
func mapMethodToAction(method string) string {
    switch method {
    case http.MethodPost:
        return "CREATE"
    case http.MethodPut, http.MethodPatch:
        return "UPDATE"
    case http.MethodDelete:
        return "DELETE"
    default:
        return "UNKNOWN"
    }
}

// extractEntityType extracts the resource type from the URL path
// Examples:
//   /v1/appointments/123 → "appointment"
//   /v1/patients/456/forms → "form"
func extractEntityType(path string) string {
    // Simplified implementation — production should use a router-based approach
    parts := strings.Split(strings.Trim(path, "/"), "/")
    if len(parts) >= 2 {
        return strings.TrimSuffix(parts[1], "s") // "appointments" → "appointment"
    }
    return "unknown"
}

// extractEntityID extracts the resource ID from the URL path
// Examples:
//   /v1/appointments/123 → "123"
//   /v1/patients/456/forms → "456" (parent resource)
func extractEntityID(path string) *int64 {
    // Simplified implementation — production should use chi.URLParam
    parts := strings.Split(strings.Trim(path, "/"), "/")
    if len(parts) >= 3 {
        if id, err := strconv.ParseInt(parts[2], 10, 64); err == nil {
            return &id
        }
    }
    return nil
}

// getClientIP resolves the real client IP in this order:
//   1. CF-Connecting-IP (Cloudflare header, trusted when App Runner only accepts CF traffic)
//   2. X-Forwarded-For (first entry in chain)
//   3. X-Real-IP (reverse proxy fallback)
//   4. r.RemoteAddr (direct connection)
func getClientIP(r *http.Request) string {
    if ip := r.Header.Get("CF-Connecting-IP"); ip != "" {
        return ip
    }
    if xff := r.Header.Get("X-Forwarded-For"); xff != "" {
        // X-Forwarded-For can be a comma-separated list: "client, proxy1, proxy2"
        // We want the first (leftmost) IP
        if idx := strings.Index(xff, ","); idx > 0 {
            return strings.TrimSpace(xff[:idx])
        }
        return xff
    }
    if ip := r.Header.Get("X-Real-IP"); ip != "" {
        return ip
    }
    return r.RemoteAddr
}

Middleware Registration

The audit middleware is registered after auth and organization context (so we have user in context), but before route handlers:

// internal/core/server/server.go

func (s *Server) registerMiddleware() {
    s.router.Use(middleware.SecurityHeaders)
    s.router.Use(middleware.AttackBlocker)
    s.router.Use(middleware.ClerkAuth(s.clerkClient))
    s.router.Use(middleware.OrganizationContext(s.adminPool, s.appPool))
    s.router.Use(middleware.SessionTimeout)
    s.router.Use(middleware.AuditLog(s.db, s.auditForwarder)) // ← Audit middleware here
}

Connection Pool vs RLS Connection

Why use the connection pool for audit writes?

The audit_log table has no RLS policies (its INSERT policy is CHECK (TRUE)). Using the RLS-scoped connection (held for the entire request) would add contention and no benefit. Instead, we use the connection pool directly:

db.Exec(ctx, ...) // Uses pool.Acquire() → quick INSERT → Release()

This keeps the RLS connection available for the main business logic query.

Error Handling

If the local audit write fails:

Log at ERROR level (triggers alerting)
DO NOT block the response (the mutation already happened)
The Telemetry forwarding still happens (best-effort)

Why not block the response?

The business logic has already executed (appointment created, patient updated, etc.)
Blocking the response would create a confusing UX (operation succeeded but response is 500)
Audit write failures are rare (database connectivity issue) and should be fixed via alerting

Monitoring:

Datadog alert: "audit: local write failed" appears in logs
Expected frequency: 0 (audit writes should never fail in normal operation)

Performance Impact

Measured latency:

Audit write: ~1ms (single INSERT, no joins, no RLS)
Total middleware overhead: ~1.5ms (includes response recording and entity extraction)

Optimization:

Use a prepared statement for the INSERT (pgx connection pool does this automatically)
Batch inserts are NOT used (we need immediate durability for HIPAA compliance)
Async forwarding to Telemetry adds no latency (non-blocking channel enqueue)

Testing

Unit Tests

// internal/core/middleware/audit_test.go

func TestAuditMiddleware(t *testing.T) {
    db := testdb.Setup(t)
    forwarder := auditevt.NewForwarder("http://telemetry.test")

    handler := AuditLog(db, forwarder)(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        w.WriteHeader(http.StatusCreated)
    }))

    req := httptest.NewRequest(http.MethodPost, "/v1/appointments", nil)
    req = req.WithContext(context.WithValue(req.Context(), userContextKey, &UserContext{
        ID:                    42,
        CurrentOrganizationID: 1,
    }))

    rec := httptest.NewRecorder()
    handler.ServeHTTP(rec, req)

    // Assert audit log was written
    var count int
    db.QueryRow(context.Background(), "SELECT COUNT(*) FROM audit_log WHERE user_id = 42").Scan(&count)
    assert.Equal(t, 1, count)
}

Integration Tests

bash

# Run integration test that verifies audit writes
go test ./internal/core/middleware -v -run TestAuditIntegration

# Expected: all mutations (POST/PUT/PATCH/DELETE) create audit entries
# Expected: GET requests do NOT create audit entries

Sensitive Data Masking

Before writing to audit_log, sensitive fields in the changes JSONB column are masked:

var sensitivePatterns = []string{
    "password", "secret", "token", "api_key", "apikey",
    "authorization", "cookie", "session",
}

func maskSensitiveFields(data map[string]any) map[string]any {
    masked := make(map[string]any)
    for key, value := range data {
        lower := strings.ToLower(key)
        isSensitive := false
        for _, pattern := range sensitivePatterns {
            if strings.Contains(lower, pattern) {
                isSensitive = true
                break
            }
        }
        if isSensitive {
            masked[key] = "[REDACTED]"
        } else {
            masked[key] = value
        }
    }
    return masked
}

Note: The current implementation does NOT populate the changes field (it's NULL in the schema). This is a placeholder for future before/after diff tracking. When implemented, all sensitive fields will be masked before writing.

Monitoring & Alerting

Metrics to track:

Audit write latency (p50, p95, p99)
Audit write error rate (should be 0%)
Audit log table size (for capacity planning)

Datadog monitors:

Alert: Audit write failures
Query: logs("audit: local write failed").rollup(count).last(5m) > 0
Severity: P1 (critical — HIPAA compliance violation risk)

Alert: Audit write latency
Query: avg:audit.write.latency.p99{env:production} > 10ms
Severity: P2 (performance degradation)

Compliance Notes

HIPAA 164.312(b) - Audit Controls:

✅ All mutations are logged
✅ Synchronous write guarantees zero event loss
✅ Append-only (no UPDATE or DELETE policies)
✅ 6-year retention (hot PostgreSQL + warm S3 archives)

GDPR Article 30 - Records of Processing:

✅ Who (user_id)
✅ What (entity_type, entity_id, changes)
✅ When (created_at)
✅ Why (action: CREATE/UPDATE/DELETE)
✅ Result (status_code)

Local Audit Logging (Core API Synchronous Write) ​

Overview ​

Why Synchronous? ​

Architecture ​

Implementation ​

Audit Middleware ​

Middleware Registration ​

Connection Pool vs RLS Connection ​

Error Handling ​

Performance Impact ​

Testing ​

Unit Tests ​

Integration Tests ​

Sensitive Data Masking ​

Monitoring & Alerting ​

Compliance Notes ​