Data Privacy and Compliance Questions

-- PostgreSQL example
BEGIN;

-- 1) Archive PII for compliance (optional)
INSERT INTO user_pii_archive (user_id, email, name, archived_at)
SELECT id, email, full_name, now()
FROM users
WHERE id = 123;

-- 2) Remove or anonymize PII in place (GDPR/DSAR)
UPDATE users
SET email = NULL, full_name = 'REDACTED', phone = NULL
WHERE id = 123;

-- 3) Delete dependent reporting/derived rows
DELETE FROM reporting_user_metrics WHERE user_id = 123;
DELETE FROM aggregated_orders WHERE user_id = 123;

-- 4) Delete child/transactional rows (orders, events)
DELETE FROM events WHERE user_id = 123;
DELETE FROM orders WHERE user_id = 123;

-- 5) Finally delete user row (or mark deleted)
DELETE FROM users WHERE id = 123;

COMMIT;

# pseudocode (python-like)
def etl_job(source, checkpoints_db, warehouse, lineage_store, audit_store):
    batch = source.next_batch()                    # includes batch_id, events[]
    if checkpoints_db.has_processed(batch.id):
        return  # idempotent skip

    try:
        checkpoints_db.mark_in_progress(batch.id, start_ts=now())

        cleaned = []
        for ev in batch.events:
            # deterministic PII stripping/pseudonymization
            ev['user_hash'] = sha256(ev.pop('email', '') + SALT)
            ev.pop('ssn', None)
            cleaned.append(ev)

        # aggregation example: daily active users per product
        agg = aggregate(cleaned, keys=['product_id','date'], metrics={'dau': count_distinct('user_hash')})

        # write with idempotent upsert using (batch_id OR aggregate_key) as conflict key
        warehouse.upsert('report_dau', agg, conflict_key=['product_id','date'])

        # record lineage: source batch -> transformations -> target tables
        lineage_store.write({
            'batch_id': batch.id,
            'rows_in': len(batch.events),
            'rows_out': len(cleaned),
            'targets': ['report_dau'],
            'transform': 'pii_strip,pseudonymize,aggregate',
            'ts': now()
        })

        checkpoints_db.mark_done(batch.id, end_ts=now())
        audit_store.write({'batch_id':batch.id,'status':'success','rows':len(batch.events),'ts':now()})
    except Exception as e:
        checkpoints_db.mark_failed(batch.id, error=str(e))
        audit_store.write({'batch_id':batch.id,'status':'failed','error':str(e),'ts':now()})
        raise  # upstream retry