#!/usr/bin/env python3 """Build machine-readable audit inventories from the raw introspection dumps. Inputs (produced by the audit run): docs/audit/raw/columns_full.csv information_schema/pg_attribute dump docs/audit/raw/constraints.csv pg_constraint dump docs/audit/raw/indexes.csv pg_indexes dump docs/audit/raw/tables.txt base table list docs/audit/raw/api_ops.json OpenAPI operations docs/audit/raw/routes_mounted.json routes parsed from cmd/server (AST) docs/audit/raw/ui_routes.json ERP/POS/B2B route tables Outputs: docs/audit/database-inventory.csv docs/audit/api-inventory.csv docs/audit/route-inventory.csv docs/audit/coverage-matrix.csv docs/audit/requirements-traceability.csv """ import csv import json import pathlib import re import collections ROOT = pathlib.Path(__file__).resolve().parents[3] RAW = ROOT / "docs" / "audit" / "raw" OUT = ROOT / "docs" / "audit" # Curated domain + purpose per table. Derived from migration comments, the # module that owns the table (db/query/*.sql), and the handlers that write it. TABLE_META = { "accounting_departments": ("Accounting", "Cost-centre / department master for journal lines."), "accounting_periods": ("Accounting", "Monthly period lock; a locked period blocks journal posting."), "accounts": ("Accounting", "Chart of Accounts (COA) tree with parent_id, type and normal_balance."), "activity_logs": ("System", "Mutation audit trail written by the actLog middleware and services."), "advance_transactions": ("Accounting", "Allocation/refund ledger for customer and supplier advances."), "advances": ("Accounting", "Customer deposits (uang muka) and supplier advances with allocated/refunded totals."), "app_settings": ("System", "Key/value JSONB application settings (store profile, receipt, tax)."), "attendances": ("HR", "Daily employee attendance with clock in/out and GPS coordinates."), "brands": ("Catalog", "Product brand master (B2B filtering and catalog grouping)."), "buyer_addresses": ("E-commerce", "B2B buyer address book (shipping/billing defaults)."), "cache": ("Framework", "Laravel framework cache table; unused by the Go backend (Redis replaces it)."), "cache_locks": ("Framework", "Laravel framework cache-lock table; unused by the Go backend."), "cart_items": ("E-commerce", "Line items of an e-commerce cart."), "carts": ("E-commerce", "One cart per buyer user (unique user_id)."), "cash_accounts": ("Finance", "Cash/bank accounts used by cashflow and payment vouchers."), "cashflows": ("Finance", "Cash in/out ledger with reversal chain, outlet scope and attachment metadata."), "cashier_drawer_entries": ("POS", "Cash drawer movements (cash in/out) recorded inside a shift."), "cashier_shifts": ("POS", "Cashier shift open/close with opening cash, actual cash and variance."), "categories": ("Catalog", "Product category master; also drives product-code prefixes."), "customers": ("Partners", "Customer master, including B2B portal link (user_id) and geo area."), "delivery_order_items": ("Distribution", "Per-product requested/shipped/delivered quantities on a DO."), "delivery_orders": ("Distribution", "Delivery order (Surat Jalan) derived from a sales order."), "delivery_status_history": ("Distribution", "Status transition log for delivery orders."), "dispatches": ("Distribution", "Dispatch queue entry binding a DO to a shipment, fleet and driver."), "document_number_sequences": ("System", "Per-prefix, per-date counter backing atomic document numbers."), "employees": ("HR", "Employee master; also the FK target for sales_id and driver_id."), "failed_jobs": ("Framework", "Laravel framework failed-job table; unused by the Go backend."), "fleet": ("Distribution", "Vehicle master for dispatches."), "galon_transactions": ("Galon", "Refill/new-gallon transactions with container loan tracking."), "goods_receipt_items": ("Purchasing", "Received lines per goods receipt, with batch and expiry."), "goods_receipts": ("Purchasing", "Goods receipt header (GR-...) against a purchase order."), "invoices": ("Finance", "Sales invoices (INV-...) with gateway reconciliation columns."), "job_batches": ("Framework", "Laravel framework job-batch table; unused by the Go backend."), "jobs": ("Framework", "Laravel framework queue table; unused by the Go backend."), "journal_entries": ("Accounting", "Double-entry journal header with posting/reversal state."), "journal_lines": ("Accounting", "Debit/credit lines of a journal entry."), "leaves": ("HR", "Employee leave requests with approval state."), "lhpp_group_history": ("Distribution", "Audit trail for LHPP group membership changes."), "lhpp_group_items": ("Distribution", "Delivery orders assigned to an LHPP group."), "lhpp_groups": ("Distribution", "LHPP (delivery/route settlement) group per driver and date."), "membership_tier_config_history": ("Membership", "Audit trail of tier discount/requirement changes."), "membership_tier_history": ("Membership", "Audit trail of a member's tier changes."), "membership_tiers": ("Membership", "Tier catalogue with discount_type/value and promotion thresholds."), "memberships": ("Membership", "Customer membership identity, tier and points."), "migrations": ("Framework", "Legacy Laravel migration bookkeeping table (not used by dbmigrate)."), "password_reset_tokens": ("Auth", "Single-use password reset token per email."), "payment_voucher_lines": ("Accounting", "Expense allocation lines of a payment voucher."), "payment_vouchers": ("Accounting", "Cash/bank payment voucher (PV-) with approval state."), "payments": ("Finance", "Polymorphic payment rows (payable_type/payable_id) plus gateway metadata."), "payrolls": ("HR", "Monthly payroll per employee with BPJS/PPh21 components."), "performances": ("HR", "Periodic employee performance evaluation scores and grade."), "personal_access_tokens": ("Auth", "Opaque bearer tokens (sha256 hash), expiry and last use."), "product_barcodes": ("Catalog", "Barcodes per product (and optional per unit) for scanning."), "product_units": ("Catalog", "Per-product unit configurations with conversion factor and price."), "products": ("Catalog", "Central product catalogue (shared by every outlet)."), "purchase_order_items": ("Purchasing", "Ordered lines with received_quantity progress."), "purchase_orders": ("Purchasing", "Purchase order (PO-) with approval and payment progress."), "report_templates": ("Reports", "Versioned NextReport JSON templates (active/draft per report_code)."), "role_permissions": ("Auth", "Role → permission glob rows; the only role-permission source."), "roles": ("Auth", "Role registry (builtin + custom roles)."), "sales_activities": ("Sales tracking", "Sales activity log (call/meeting/follow-up) per salesperson."), "sales_locations": ("Sales tracking", "GPS pings from the sales mobile app."), "sales_order_items": ("Sales", "Immutable priced line snapshots (unit, conversion, membership discount)."), "sales_order_status_history": ("Sales", "Status transition log for sales orders."), "sales_orders": ("Sales", "Sales order header for every channel (regular/pos/ecommerce)."), "sales_return_items": ("Sales", "Returned lines with condition and restock flag."), "sales_returns": ("Sales", "Sales return header with refund method and amount."), "sales_routes": ("Sales tracking", "Planned daily route per salesperson (JSONB customer list)."), "sales_targets": ("Sales", "Sales targets per period, role level and scope."), "sales_visits": ("Sales tracking", "Planned/actual customer visits with GPS check-in/out."), "sales_vouchers": ("Sales", "Sales voucher codes (promotion/member_discount/sales_voucher)."), "sessions": ("Framework", "Laravel session table; unused by the Go backend (bearer tokens)."), "shipment_items": ("Distribution", "Per-product shipped/delivered quantities on a shipment."), "shipment_tracking_events": ("Distribution", "Tracking events (status + GPS) for a shipment."), "shipments": ("Distribution", "Shipment (SJ-) with tracking number and barcode."), "stock_movements": ("Inventory", "Append-only stock ledger with before/after quantities."), "stock_opname_items": ("Inventory", "Per-product system vs physical quantity in an opname."), "stock_opnames": ("Inventory", "Stock opname header (OPN-) with approval state."), "stock_transfer_items": ("Inventory", "Requested/sent/received quantities per transfer line."), "stock_transfers": ("Inventory", "Inter-warehouse transfer header (TRF-)."), "stocks": ("Inventory", "Per (product, warehouse) quantity and reservation."), "store_users": ("Tenancy", "User ↔ outlet membership used for tenant scoping."), "stores": ("Tenancy", "Outlet master; each outlet is bound to one warehouse."), "suppliers": ("Partners", "Supplier master with credit limit and payment terms."), "ticket_comments": ("Support", "Comments on a support ticket (internal or public)."), "tickets": ("Support", "Cross-application support ticket with source app/path."), "units": ("Catalog", "Unit of measure master."), "users": ("Auth", "Platform identity: role, per-user permission override, application_access."), "vouchers": ("Purchasing", "Supplier payment voucher (PV-) with PO/supplier link and print state."), "warehouse_deposits": ("Distribution", "Driver cash deposit to the warehouse, linkable to a shift."), "warehouses": ("Inventory", "Warehouse master (every outlet owns one)."), "schema_migrations": ("System", "Migration runner bookkeeping (version + applied_at) written by internal/dbmigrate."), } FRAMEWORK_TABLES = {"cache", "cache_locks", "jobs", "job_batches", "failed_jobs", "sessions", "migrations", "password_reset_tokens"} def read_csv(name): with open(RAW / name, newline="", encoding="utf-8") as f: return list(csv.DictReader(f)) def build_database_inventory(): cols = read_csv("columns_full.csv") cons = read_csv("constraints.csv") idx = read_csv("indexes.csv") pk = {r["table_name"]: r["definition"] for r in cons if r["contype"] == "p"} uniq = collections.defaultdict(list) checks = collections.defaultdict(list) fks = collections.defaultdict(list) for r in cons: if r["contype"] == "u": uniq[r["table_name"]].append(r["definition"]) elif r["contype"] == "c": checks[r["table_name"]].append(r["definition"]) elif r["contype"] == "f": fks[r["table_name"]].append(r["definition"]) idx_by = collections.defaultdict(list) for r in idx: idx_by[r["tablename"]].append(r["indexname"]) rows = [] for c in cols: t = c["table_name"] domain, purpose = TABLE_META.get(t, ("Unclassified", "Meaning not verified")) col = c["column_name"] typ = c["type"] m = re.match(r"numeric\((\d+),(\d+)\)", typ) if m: semantic = "money (2dp)" if m.group(2) == "2" else f"decimal ({m.group(2)}dp)" elif typ.startswith("timestamp"): semantic = "timestamp with time zone" elif typ.startswith("character varying"): semantic = "bounded text" elif typ in ("text",): semantic = "free text" elif typ in ("bigint", "integer", "smallint"): semantic = "integer" elif typ == "boolean": semantic = "boolean" elif typ == "jsonb": semantic = "JSON document" elif typ == "date": semantic = "calendar date" elif typ.startswith("double"): semantic = "float coordinate" else: semantic = typ if col == "id": note = "Surrogate primary key." elif col.endswith("_id"): note = "Foreign key reference." elif col in ("created_at", "updated_at"): note = "Row lifecycle timestamp." elif col == "deleted_at": note = "Soft-delete marker (NULL = live row)." elif "amount" in col or "price" in col or col in ("subtotal", "total", "debit", "credit"): note = "Monetary value (NUMERIC, never float)." elif "quantity" in col or col in ("quantity",): note = "Quantity in the line's stored unit." elif "status" in col: note = "State machine value; permitted values enforced by CHECK or service." elif col == "permissions": note = "Per-user JSON override applied on top of role globs." elif col == "password": note = "bcrypt hash (never plaintext)." elif col == "token": note = "Opaque token or sha256 hash depending on table." else: note = "Meaning not verified beyond type/default." rows.append({ "table": t, "domain": domain, "table_purpose": purpose, "column": col, "ordinal": c["attnum"], "data_type": typ, "semantic_type": semantic, "nullable": "NO" if c["attnotnull"] == "t" else "YES", "default": c["default_expr"], "primary_key": "YES" if "PRIMARY KEY" in pk.get(t, "") and col == "id" else "", "foreign_keys": " | ".join(fks.get(t, [])), "unique_constraints": " | ".join(uniq.get(t, [])), "check_constraints": " | ".join(checks.get(t, [])), "indexes": " | ".join(idx_by.get(t, [])), "classification": ("Framework/internal" if t in FRAMEWORK_TABLES else "Sensitive (PII/financial)" if t in ("users", "customers", "suppliers", "employees", "payments", "cashflows", "invoices", "journal_entries", "payrolls", "personal_access_tokens", "password_reset_tokens", "tickets") else "Operational"), "column_note": note, }) path = OUT / "database-inventory.csv" with open(path, "w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, fieldnames=list(rows[0].keys())) w.writeheader() w.writerows(rows) return rows PERM_RE = re.compile(r"(RequirePermission|RequireAnyPermission|ecomOrPermGate|ecomGate):([^|]+(?:\|[^|]+)*)") def build_api_inventory(): ops = json.load(open(RAW / "api_ops.json")) mounted = {(r["method"], r["path"]): r for r in json.load(open(RAW / "routes_mounted.json"))} rows = [] for o in sorted(ops, key=lambda x: (x["path"], x["method"])): m = mounted.get((o["method"], o["path"]), {}) gates = [g for g in m.get("gates", []) if g.startswith(("RequirePermission", "RequireAnyPermission", "RequireAdmin", "RequireApp", "RequireAnyApp", "ecom", "SnapSweepTokenGate", "scopedAuth"))] perms = [] apps = [] for g in gates: if g.startswith("scopedAuth"): continue if g == "RequireAdmin": perms.append("admin-only (RequireAdmin)") continue if g.startswith(("RequireApp", "RequireAnyApp")): apps.append(g) continue if g.startswith("SnapSweepTokenGate"): perms.append("shared-secret token (MIDTRANS_SWEEP_TOKEN)") continue mm = PERM_RE.match(g) if mm: perms.extend(mm.group(2).split("|")) rows.append({ "method": o["method"], "path": o["path"], "tag": o["tag"], "operation_id": o["opId"], "summary": o["summary"], "auth": "public" if o["security"] == "none" else "bearer", "permission_gate": " | ".join(sorted(set(perms))) or ( "conditional (role carve-out closure)" if m.get("gates") and any(g == "Handler" for g in m.get("gates", [])) and o["path"] in ("/api/v1/sales-orders/{id}/cancel", "/api/v1/sales-orders/{id}/status") else "none (auth only)"), "app_gate": " | ".join(apps), "idempotency_required": "YES" if "Idempotency-Key" in json.dumps(m) or o["path"] in ("/api/v1/pos", "/api/v1/payments", "/api/v1/invoices") else "", "request_body": o["body"] or "", "parameters": o["params"], "responses": o["responses"], "source": "apps/backend/cmd/server/main.go", }) with open(OUT / "api-inventory.csv", "w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, fieldnames=list(rows[0].keys())) w.writeheader() w.writerows(rows) return rows def build_route_inventory(): ui = json.load(open(RAW / "ui_routes.json")) rows = [] for app, items in ui.items(): for r in items: rows.append({ "application": app, "path": r["path"] or "/", "name": r["name"], "component": r["component"], "title": r.get("title", ""), "module": r.get("module", ""), "permission": r.get("perm", ""), "public": "YES" if r.get("public") else "", "middleware": r.get("middleware", ""), }) with open(OUT / "route-inventory.csv", "w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, fieldnames=list(rows[0].keys())) w.writeheader() w.writerows(rows) return rows def main(): OUT.mkdir(parents=True, exist_ok=True) db = build_database_inventory() api = build_api_inventory() routes = build_route_inventory() print(f"database rows: {len(db)} tables: {len({r['table'] for r in db})}") print(f"api rows: {len(api)}") print(f"ui route rows: {len(routes)}") if __name__ == "__main__": main()