README.md
# Treasury Operations Demo
This workspace contains a fictional multi-service treasury operations portal for local benchmarking only.
Services:
- `portal/` - TypeScript/JavaScript admin portal built on Express
- `payments-api/` - Go API for transfers and settlement jobs
- `docs-worker/` - Python Flask service for documents and background work
- `risk-api/` - Java HTTP API for risk operations
Each service uses local persistence and exposes a few thin HTTP routes with middleware, audit logging, and basic RBAC checks.
docs-worker/app.py
from __future__ import annotations
import json
import os
import queue
import sqlite3
import threading
from datetime import datetime
from pathlib import Path
from flask import Flask, Response, g, jsonify, request, send_file
BASE_DIR = Path(__file__).resolve().parent
DATA_DIR = BASE_DIR / "data"
STORAGE_DIR = DATA_DIR / "storage"
DB_PATH = DATA_DIR / "docs.db"
DATA_DIR.mkdir(parents=True, exist_ok=True)
STORAGE_DIR.mkdir(parents=True, exist_ok=True)
app = Flask(__name__)
job_queue: "queue.Queue[dict]" = queue.Queue()
def get_db() -> sqlite3.Connection:
if "db" not in g:
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
g.db = conn
return g.db
def init_db() -> None:
db = sqlite3.connect(DB_PATH)
db.executescript(
"""
create table if not exists organizations (
id integer primary key autoincrement,
name text not null,
org_code text not null unique,
region text not null,
tier text not null
);
create table if not exists teams (
id integer primary key autoincrement,
org_id integer not null,
name text not null,
cost_center text not null,
status text not null default 'active'
);
create table if not exists users (
id integer primary key autoincrement,
org_id integer not null,
team_id integer,
email text not null,
display_name text not null,
role text not null,
status text not null default 'active'
);
create table if not exists accounts (
id integer primary key autoincrement,
org_id integer not null,
account_no text not null,
currency text not null,
balance numeric not null default 0
);
create table if not exists beneficiaries (
id integer primary key autoincrement,
org_id integer not null,
name text not null,
bank_name text not null,
routing text not null
);
create table if not exists transfers (
id integer primary key autoincrement,
org_id integer not null,
account_id integer not null,
beneficiary_id integer not null,
amount numeric not null,
currency text not null,
status text not null,
memo text not null,
requested_by text not null
);
create table if not exists documents (
id integer primary key autoincrement,
org_id integer not null,
filename text not null,
original_name text not null,
mime_type text not null,
kind text not null,
created_at text not null default current_timestamp
);
create table if not exists risk_reviews (
id integer primary key autoincrement,
org_id integer not null,
subject text not null,
score numeric not null,
note text not null,
created_at text not null default current_timestamp
);
create table if not exists audit_records (
id integer primary key autoincrement,
org_id integer,
actor text not null,
action text not null,
resource text not null,
details text not null,
created_at text not null default current_timestamp
);
"""
)
if db.execute("select count(*) from organizations").fetchone()[0] == 0:
db.execute(
"insert into organizations (name, org_code, region, tier) values (?, ?, ?, ?)",
("Acme Treasury", "acme", "us-east", "enterprise"),
)
db.execute(
"insert into organizations (name, org_code, region, tier) values (?, ?, ?, ?)",
("Northwind Holdings", "northwind", "eu-west", "growth"),
)
db.commit()
db.close()
def current_identity() -> dict:
return {
"org_id": int(request.headers.get("X-Org-Id", "1")),
"user_id": request.headers.get("X-User-Id", "docs-user"),
"role": request.headers.get("X-Role", "viewer"),
}
@app.before_request
def load_identity() -> None:
g.identity = current_identity()
g.started_at = datetime.utcnow().isoformat()
@app.after_request
def audit_request(response: Response) -> Response:
identity = getattr(g, "identity", None)
if identity:
db = get_db()
db.execute(
"insert into audit_records (org_id, actor, action, resource, details) values (?, ?, ?, ?, ?)",
(
identity["org_id"],
identity["user_id"],
f"{request.method} {request.path}",
request.full_path,
json.dumps({"status": response.status_code, "args": request.args.to_dict(), "json": request.get_json(silent=True)}),
),
)
db.commit()
return response
@app.teardown_appcontext
def close_db(exc: Exception | None) -> None:
db = g.pop("db", None)
if db is not None:
db.close()
def require_roles(*allowed: str):
def decorator(fn):
def wrapped(*args, **kwargs):
if g.identity["role"] not in allowed:
return jsonify({"error": "forbidden"}), 403
return fn(*args, **kwargs)
wrapped.__name__ = fn.__name__
return wrapped
return decorator
def render_report(org_name: str, title: str, notes: str) -> str:
return f"""
<html>
<body>
<h1>{title}</h1>
<h2>{org_name}</h2>
<div>{notes}</div>
</body>
</html>
"""
def enqueue_job(payload: dict) -> None:
job_queue.put(payload)
def job_worker() -> None:
while True:
payload = job_queue.get()
if payload is None:
break
db = sqlite3.connect(DB_PATH)
db.execute(
"insert into audit_records (org_id, actor, action, resource, details) values (?, ?, ?, ?, ?)",
(
payload.get("org_id"),
"system",
"job.dispatch",
payload.get("kind", "document"),
json.dumps(payload),
),
)
if payload.get("kind") == "document":
filename = payload["filename"]
target = STORAGE_DIR / filename
target.write_text(payload.get("content", ""), encoding="utf-8")
db.execute(
"insert into documents (org_id, filename, original_name, mime_type, kind) values (?, ?, ?, ?, ?)",
(
payload.get("org_id"),
filename,
payload.get("original_name", filename),
payload.get("mime_type", "text/plain"),
payload.get("kind", "document"),
),
)
db.commit()
db.close()
job_queue.task_done()
init_db()
threading.Thread(target=job_worker, daemon=True).start()
@app.route("/health", methods=["GET"])
def health() -> tuple:
db = get_db()
counts = {
"documents": db.execute("select count(*) from documents").fetchone()[0],
"audit_records": db.execute("select count(*) from audit_records").fetchone()[0],
"jobs_pending": job_queue.qsize(),
}
return jsonify({"ok": True, "counts": counts})
@app.route("/documents", methods=["GET"])
@require_roles("admin", "manager", "viewer")
def search_documents() -> tuple:
clauses = ["org_id = ?"]
params: list[object] = [g.identity["org_id"]]
kind = request.args.get("kind", "").strip()
name = request.args.get("name", "").strip()
if kind:
clauses.append("kind = ?")
params.append(kind)
if name:
clauses.append("original_name like ?")
params.append(f"%{name}%")
rows = get_db().execute(
f"select id, filename, original_name, mime_type, kind, created_at from documents where {' and '.join(clauses)} order by id desc",
params,
).fetchall()
return jsonify({"data": [dict(row) for row in rows]})
@app.route("/docs/render", methods=["POST"])
@require_roles("admin", "manager")
def render_document() -> tuple:
payload = request.get_json(force=True)
org_name = payload["orgName"]
title = payload.get("title", "Treasury Notice")
notes = payload.get("notes", "")
filename = payload.get("filename", f"report-{int(datetime.utcnow().timestamp())}.html")
html = render_report(org_name, title, notes)
target = STORAGE_DIR / filename
target.write_text(html, encoding="utf-8")
db = get_db()
db.execute(
"insert into documents (org_id, filename, original_name, mime_type, kind) values (?, ?, ?, ?, ?)",
(g.identity["org_id"], filename, filename, "text/html", "report"),
)
db.commit()
return jsonify({"filename": filename, "path": str(target)}), 201
@app.route("/docs/upload", methods=["POST"])
@require_roles("admin", "manager")
def upload_document() -> tuple:
upload = request.files.get("file")
if upload is None:
return jsonify({"error": "missing file"}), 400
filename = f"{int(datetime.utcnow().timestamp())}-{upload.filename}"
target = STORAGE_DIR / filename
upload.save(target)
db = get_db()
db.execute(
"insert into documents (org_id, filename, original_name, mime_type, kind) values (?, ?, ?, ?, ?)",
(g.identity["org_id"], filename, upload.filename, upload.mimetype or "application/octet-stream", "upload"),
)
db.commit()
return jsonify({"filename": filename}), 201
@app.route("/docs/download/<path:filename>", methods=["GET"])
@require_roles("admin", "manager", "viewer")
def download_document(filename: str):
target = STORAGE_DIR / filename
if not target.exists():
return jsonify({"error": "not found"}), 404
return send_file(target, as_attachment=True)
@app.route("/jobs/dispatch", methods=["POST"])
@require_roles("admin", "manager")
def dispatch_job() -> tuple:
payload = request.get_json(force=True)
payload["org_id"] = g.identity["org_id"]
enqueue_job(payload)
return jsonify({"queued": True, "kind": payload.get("kind", "document")}), 202
@app.route("/reports/summary", methods=["GET"])
@require_roles("admin", "manager")
def summary_report() -> tuple:
db = get_db()
org_id = int(request.args.get("orgId", g.identity["org_id"]))
title = request.args.get("title", "Document Summary")
org = db.execute("select name, region, tier from organizations where id = ?", (org_id,)).fetchone()
if org is None:
return jsonify({"error": "missing org"}), 404
notes = request.args.get("notes", "No notes supplied")
html = render_report(org["name"], title, notes)
return Response(html, mimetype="text/html")
if __name__ == "__main__":
app.run(host="0.0.0.0", port=int(os.environ.get("PORT", "4300")), debug=False)
docs-worker/requirements.txt
Flask==3.0.3
payments-api/go.mod
module payments-api
go 1.22
require modernc.org/sqlite v1.36.0
require (
github.com/dustin/go-humanize v1.0.1 // indirect
github.com/google/uuid v1.6.0 // indirect
github.com/mattn/go-isatty v0.0.20 // indirect
github.com/ncruces/go-strftime v0.1.9 // indirect
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
golang.org/x/exp v0.0.0-20230315142452-642cacee5cc0 // indirect
golang.org/x/sys v0.30.0 // indirect
modernc.org/libc v1.61.13 // indirect
modernc.org/mathutil v1.7.1 // indirect
modernc.org/memory v1.8.2 // indirect
)
payments-api/go.sum
github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
github.com/google/pprof v0.0.0-20240409012703-83162a5b38cd h1:gbpYu9NMq8jhDVbvlGkMFWCjLFlqqEZjEmObmhUy6Vo=
github.com/google/pprof v0.0.0-20240409012703-83162a5b38cd/go.mod h1:kf6iHlnVGwgKolg33glAes7Yg/8iWP8ukqeldJSO7jw=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
github.com/ncruces/go-strftime v0.1.9 h1:bY0MQC28UADQmHmaF5dgpLmImcShSi2kHU9XLdhx/f4=
github.com/ncruces/go-strftime v0.1.9/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls=
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
golang.org/x/exp v0.0.0-20230315142452-642cacee5cc0 h1:pVgRXcIictcr+lBQIFeiwuwtDIs4eL21OuM9nyAADmo=
golang.org/x/exp v0.0.0-20230315142452-642cacee5cc0/go.mod h1:CxIveKay+FTh1D0yPZemJVgC/95VzuuOLq5Qi4xnoYc=
golang.org/x/mod v0.19.0 h1:fEdghXQSo20giMthA7cd28ZC+jts4amQ3YMXiP5oMQ8=
golang.org/x/mod v0.19.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M=
golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc=
golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/tools v0.23.0 h1:SGsXPZ+2l4JsgaCKkx+FQ9YZ5XEtA1GZYuoDjenLjvg=
golang.org/x/tools v0.23.0/go.mod h1:pnu6ufv6vQkll6szChhK3C3L/ruaIv5eBeztNG8wtsI=
modernc.org/cc/v4 v4.24.4 h1:TFkx1s6dCkQpd6dKurBNmpo+G8Zl4Sq/ztJ+2+DEsh0=
modernc.org/cc/v4 v4.24.4/go.mod h1:uVtb5OGqUKpoLWhqwNQo/8LwvoiEBLvZXIQ/SmO6mL0=
modernc.org/ccgo/v4 v4.23.16 h1:Z2N+kk38b7SfySC1ZkpGLN2vthNJP1+ZzGZIlH7uBxo=
modernc.org/ccgo/v4 v4.23.16/go.mod h1:nNma8goMTY7aQZQNTyN9AIoJfxav4nvTnvKThAeMDdo=
modernc.org/fileutil v1.3.0 h1:gQ5SIzK3H9kdfai/5x41oQiKValumqNTDXMvKo62HvE=
modernc.org/fileutil v1.3.0/go.mod h1:XatxS8fZi3pS8/hKG2GH/ArUogfxjpEKs3Ku3aK4JyQ=
modernc.org/gc/v2 v2.6.3 h1:aJVhcqAte49LF+mGveZ5KPlsp4tdGdAOT4sipJXADjw=
modernc.org/gc/v2 v2.6.3/go.mod h1:YgIahr1ypgfe7chRuJi2gD7DBQiKSLMPgBQe9oIiito=
modernc.org/libc v1.61.13 h1:3LRd6ZO1ezsFiX1y+bHd1ipyEHIJKvuprv0sLTBwLW8=
modernc.org/libc v1.61.13/go.mod h1:8F/uJWL/3nNil0Lgt1Dpz+GgkApWh04N3el3hxJcA6E=
modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU=
modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg=
modernc.org/memory v1.8.2 h1:cL9L4bcoAObu4NkxOlKWBWtNHIsnnACGF/TbqQ6sbcI=
modernc.org/memory v1.8.2/go.mod h1:ZbjSvMO5NQ1A2i3bWeDiVMxIorXwdClKE/0SZ+BMotU=
modernc.org/opt v0.1.4 h1:2kNGMRiUjrp4LcaPuLY2PzUfqM/w9N23quVwhKt5Qm8=
modernc.org/opt v0.1.4/go.mod h1:03fq9lsNfvkYSfxrfUhZCWPk1lm4cq4N+Bh//bEtgns=
modernc.org/sortutil v1.2.1 h1:+xyoGf15mM3NMlPDnFqrteY07klSFxLElE2PVuWIJ7w=
modernc.org/sortutil v1.2.1/go.mod h1:7ZI3a3REbai7gzCLcotuw9AC4VZVpYMjDzETGsSMqJE=
modernc.org/sqlite v1.36.0 h1:EQXNRn4nIS+gfsKeUTymHIz1waxuv5BzU7558dHSfH8=
modernc.org/sqlite v1.36.0/go.mod h1:7MPwH7Z6bREicF9ZVUR78P1IKuxfZ8mRIDHD0iD+8TU=
modernc.org/strutil v1.2.1 h1:UneZBkQA+DX2Rp35KcM69cSsNES9ly8mQWD71HKlOA0=
modernc.org/strutil v1.2.1/go.mod h1:EHkiggD70koQxjVdSBM3JKM7k6L0FbGE5eymy9i3B9A=
modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y=
modernc.org/token v1.1.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM=
payments-api/main.go
package main
import (
"context"
"database/sql"
"encoding/json"
"fmt"
"log"
"net/http"
"os"
"path/filepath"
"strconv"
"strings"
"sync"
_ "modernc.org/sqlite"
)
type identity struct {
OrgID int64
UserID string
Role string
}
type ctxKey string
const identityKey ctxKey = "identity"
type job struct {
TransferID int64
OrgID int64
}
type responseRecorder struct {
http.ResponseWriter
status int
}
func (r *responseRecorder) WriteHeader(code int) {
r.status = code
r.ResponseWriter.WriteHeader(code)
}
var (
db *sql.DB
jobs = make(chan job, 64)
startOnce sync.Once
dataDir = filepath.Join("data")
dbPath = filepath.Join(dataDir, "payments.db")
)
func main() {
mustInit()
startWorker()
http.HandleFunc("/healthz", withMiddleware(healthHandler))
http.HandleFunc("/accounts/search", withMiddleware(requireRoles("admin", "manager", "viewer")(accountsSearchHandler)))
http.HandleFunc("/beneficiaries", withMiddleware(requireRoles("admin", "manager", "viewer")(beneficiariesHandler)))
http.HandleFunc("/transfers", withMiddleware(requireRoles("admin", "manager")(transfersHandler)))
http.HandleFunc("/jobs/settlements", withMiddleware(requireRoles("admin", "manager")(settlementHandler)))
http.HandleFunc("/audit", withMiddleware(requireRoles("admin")(auditHandler)))
port := getenv("PORT", "4200")
log.Printf("payments-api listening on %s", port)
log.Fatal(http.ListenAndServe(":"+port, nil))
}
func mustInit() {
if err := os.MkdirAll(dataDir, 0o755); err != nil {
log.Fatal(err)
}
var err error
db, err = sql.Open("sqlite", "file:"+dbPath+"?_pragma=busy_timeout(5000)")
if err != nil {
log.Fatal(err)
}
db.SetMaxOpenConns(1)
for _, stmt := range []string{
`create table if not exists organizations (
id integer primary key autoincrement,
name text not null,
org_code text not null unique,
region text not null,
tier text not null
);`,
`create table if not exists teams (
id integer primary key autoincrement,
org_id integer not null,
name text not null,
cost_center text not null,
status text not null
);`,
`create table if not exists users (
id integer primary key autoincrement,
org_id integer not null,
team_id integer,
email text not null,
display_name text not null,
role text not null
);`,
`create table if not exists accounts (
id integer primary key autoincrement,
org_id integer not null,
account_no text not null,
currency text not null,
balance numeric not null default 0
);`,
`create table if not exists beneficiaries (
id integer primary key autoincrement,
org_id integer not null,
name text not null,
bank_name text not null,
routing text not null
);`,
`create table if not exists transfers (
id integer primary key autoincrement,
org_id integer not null,
account_id integer not null,
beneficiary_id integer not null,
amount numeric not null,
currency text not null,
status text not null,
memo text not null,
requested_by text not null,
created_at text not null default current_timestamp
);`,
`create table if not exists audit_records (
id integer primary key autoincrement,
org_id integer,
actor text not null,
action text not null,
resource text not null,
details text not null,
created_at text not null default current_timestamp
);`,
} {
if _, err := db.Exec(stmt); err != nil {
log.Fatal(err)
}
}
seedData()
}
func seedData() {
var count int
_ = db.QueryRow("select count(*) from organizations").Scan(&count)
if count > 0 {
return
}
orgs := []struct {
name, code, region, tier string
}{
{"Acme Treasury", "acme", "us-east", "enterprise"},
{"Northwind Holdings", "northwind", "eu-west", "growth"},
}
for _, org := range orgs {
res, _ := db.Exec("insert into organizations (name, org_code, region, tier) values (?, ?, ?, ?)", org.name, org.code, org.region, org.tier)
orgID, _ := res.LastInsertId()
_, _ = db.Exec("insert into teams (org_id, name, cost_center, status) values (?, ?, ?, ?)", orgID, "Operations", "100-ops", "active")
_, _ = db.Exec("insert into teams (org_id, name, cost_center, status) values (?, ?, ?, ?)", orgID, "Payments", "200-pay", "active")
_, _ = db.Exec("insert into accounts (org_id, account_no, currency, balance) values (?, ?, ?, ?)", orgID, "ACCT-"+strconv.FormatInt(orgID, 10), "USD", 1000000)
_, _ = db.Exec("insert into beneficiaries (org_id, name, bank_name, routing) values (?, ?, ?, ?)", orgID, "Primary Vendor", "Metro Trust", "110000")
}
}
func startWorker() {
startOnce.Do(func() {
go func() {
for j := range jobs {
_, _ = db.Exec("update transfers set status = ? where id = ? and org_id = ?", "settled", j.TransferID, j.OrgID)
audit(j.OrgID, "system", "settlement", fmt.Sprintf("transfer:%d", j.TransferID), map[string]any{"status": "settled"})
}
}()
})
}
func withMiddleware(next http.HandlerFunc) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
rec := &responseRecorder{ResponseWriter: w, status: http.StatusOK}
id := identityFromRequest(r)
r = r.WithContext(context.WithValue(r.Context(), identityKey, id))
next(rec, r)
audit(id.OrgID, id.UserID, r.Method+" "+r.URL.Path, r.URL.String(), map[string]any{
"status": rec.status,
"query": r.URL.Query(),
})
}
}
func identityFromRequest(r *http.Request) identity {
orgID, _ := strconv.ParseInt(r.Header.Get("X-Org-Id"), 10, 64)
if orgID == 0 {
orgID = 1
}
role := r.Header.Get("X-Role")
if role == "" {
role = "viewer"
}
userID := r.Header.Get("X-User-Id")
if userID == "" {
userID = "payments-user"
}
return identity{OrgID: orgID, UserID: userID, Role: role}
}
func requireRoles(allowed ...string) func(http.HandlerFunc) http.HandlerFunc {
return func(next http.HandlerFunc) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
id := mustIdentity(r)
for _, role := range allowed {
if id.Role == role {
next(w, r)
return
}
}
http.Error(w, "forbidden", http.StatusForbidden)
}
}
}
func mustIdentity(r *http.Request) identity {
if v := r.Context().Value(identityKey); v != nil {
if id, ok := v.(identity); ok {
return id
}
}
return identityFromRequest(r)
}
func healthHandler(w http.ResponseWriter, r *http.Request) {
var orgCount, transferCount int
_ = db.QueryRow("select count(*) from organizations").Scan(&orgCount)
_ = db.QueryRow("select count(*) from transfers").Scan(&transferCount)
_ = json.NewEncoder(w).Encode(map[string]any{
"ok": true,
"organizations": orgCount,
"transfers": transferCount,
})
}
func accountsSearchHandler(w http.ResponseWriter, r *http.Request) {
id := mustIdentity(r)
clauses := []string{"org_id = ?"}
args := []any{id.OrgID}
if currency := strings.TrimSpace(r.URL.Query().Get("currency")); currency != "" {
clauses = append(clauses, "currency = ?")
args = append(args, currency)
}
if accountNo := strings.TrimSpace(r.URL.Query().Get("accountNo")); accountNo != "" {
clauses = append(clauses, "account_no like ?")
args = append(args, "%"+accountNo+"%")
}
query := "select id, account_no, currency, balance from accounts where " + strings.Join(clauses, " and ") + " order by id desc"
rows, err := db.Query(query, args...)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
defer rows.Close()
var data []map[string]any
for rows.Next() {
var accountID int64
var accountNo, currency string
var balance float64
_ = rows.Scan(&accountID, &accountNo, ¤cy, &balance)
data = append(data, map[string]any{
"id": accountID,
"accountNo": accountNo,
"currency": currency,
"balance": balance,
})
}
_ = json.NewEncoder(w).Encode(map[string]any{"data": data})
}
func beneficiariesHandler(w http.ResponseWriter, r *http.Request) {
id := mustIdentity(r)
switch r.Method {
case http.MethodPost:
var body struct {
Name string `json:"name"`
BankName string `json:"bankName"`
Routing string `json:"routing"`
}
_ = json.NewDecoder(r.Body).Decode(&body)
res, err := db.Exec("insert into beneficiaries (org_id, name, bank_name, routing) values (?, ?, ?, ?)", id.OrgID, body.Name, body.BankName, body.Routing)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
insertedID, _ := res.LastInsertId()
_ = json.NewEncoder(w).Encode(map[string]any{"id": insertedID, "name": body.Name})
default:
nameQuery := strings.TrimSpace(r.URL.Query().Get("name"))
query := "select id, name, bank_name, routing from beneficiaries where org_id = ?"
args := []any{id.OrgID}
if nameQuery != "" {
query += " and name like ?"
args = append(args, "%"+nameQuery+"%")
}
rows, err := db.Query(query+" order by id desc", args...)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
defer rows.Close()
var data []map[string]any
for rows.Next() {
var beneficiaryID int64
var name, bankName, routing string
_ = rows.Scan(&beneficiaryID, &name, &bankName, &routing)
data = append(data, map[string]any{
"id": beneficiaryID,
"name": name,
"bankName": bankName,
"routing": routing,
})
}
_ = json.NewEncoder(w).Encode(map[string]any{"data": data})
}
}
func transfersHandler(w http.ResponseWriter, r *http.Request) {
id := mustIdentity(r)
switch r.Method {
case http.MethodPost:
var body struct {
AccountID int64 `json:"accountId"`
BeneficiaryID int64 `json:"beneficiaryId"`
Amount float64 `json:"amount"`
Currency string `json:"currency"`
Memo string `json:"memo"`
}
_ = json.NewDecoder(r.Body).Decode(&body)
res, err := db.Exec(
"insert into transfers (org_id, account_id, beneficiary_id, amount, currency, status, memo, requested_by) values (?, ?, ?, ?, ?, ?, ?, ?)",
id.OrgID, body.AccountID, body.BeneficiaryID, body.Amount, body.Currency, "queued", body.Memo, id.UserID,
)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
transferID, _ := res.LastInsertId()
jobs <- job{TransferID: transferID, OrgID: id.OrgID}
_ = json.NewEncoder(w).Encode(map[string]any{"id": transferID, "status": "queued"})
default:
rows, err := db.Query("select id, account_id, beneficiary_id, amount, currency, status, memo from transfers where org_id = ? order by id desc limit 50", id.OrgID)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
defer rows.Close()
var data []map[string]any
for rows.Next() {
var transferID, accountID, beneficiaryID int64
var amount float64
var currency, status, memo string
_ = rows.Scan(&transferID, &accountID, &beneficiaryID, &amount, ¤cy, &status, &memo)
data = append(data, map[string]any{
"id": transferID,
"accountId": accountID,
"beneficiaryId": beneficiaryID,
"amount": amount,
"currency": currency,
"status": status,
"memo": memo,
})
}
_ = json.NewEncoder(w).Encode(map[string]any{"data": data})
}
}
func settlementHandler(w http.ResponseWriter, r *http.Request) {
id := mustIdentity(r)
if r.Method != http.MethodPost {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
var body struct {
TransferID int64 `json:"transferId"`
}
_ = json.NewDecoder(r.Body).Decode(&body)
jobs <- job{TransferID: body.TransferID, OrgID: id.OrgID}
_ = json.NewEncoder(w).Encode(map[string]any{"queued": true, "transferId": body.TransferID})
}
func auditHandler(w http.ResponseWriter, r *http.Request) {
id := mustIdentity(r)
rows, err := db.Query("select id, actor, action, resource, details, created_at from audit_records where org_id = ? order by id desc limit 100", id.OrgID)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
defer rows.Close()
var data []map[string]any
for rows.Next() {
var auditID int64
var actor, action, resource, details, createdAt string
_ = rows.Scan(&auditID, &actor, &action, &resource, &details, &createdAt)
data = append(data, map[string]any{
"id": auditID, "actor": actor, "action": action, "resource": resource, "details": details, "createdAt": createdAt,
})
}
_ = json.NewEncoder(w).Encode(map[string]any{"data": data})
}
func audit(orgID int64, actor, action, resource string, details map[string]any) {
_, _ = db.Exec(
"insert into audit_records (org_id, actor, action, resource, details) values (?, ?, ?, ?, ?)",
orgID, actor, action, resource, mustJSON(details),
)
}
func mustJSON(v any) string {
b, _ := json.Marshal(v)
return string(b)
}
func getenv(key, fallback string) string {
if value := os.Getenv(key); value != "" {
return value
}
return fallback
}
portal/package.json
{
"name": "treasury-portal",
"private": true,
"version": "1.0.0",
"type": "commonjs",
"scripts": {
"dev": "ts-node src/server.ts",
"build": "tsc -p tsconfig.json",
"start": "node dist/server.js"
},
"dependencies": {
"better-sqlite3": "^11.10.0",
"express": "^4.21.2",
"multer": "^1.4.5-lts.1"
},
"devDependencies": {
"@types/express": "^5.0.3",
"@types/multer": "^1.4.12",
"@types/node": "^22.15.3",
"ts-node": "^10.9.2",
"typescript": "^5.8.3"
}
}
portal/src/server.ts
import express, { NextFunction, Request, Response } from "express";
import fs from "fs";
import path from "path";
import multer from "multer";
import { db, recordAudit, storagePath } from "./store.js";
type Identity = {
orgId: number;
userId: string;
role: "admin" | "manager" | "viewer";
};
declare global {
namespace Express {
interface Request {
identity?: Identity;
}
}
}
const app = express();
const upload = multer({ dest: path.resolve(__dirname, "..", "tmp") });
fs.mkdirSync(path.resolve(__dirname, "..", "tmp"), { recursive: true });
app.use(express.json());
app.use(express.urlencoded({ extended: true }));
app.use((req, res, next) => {
const orgId = Number(req.header("x-org-id") || "1");
const userId = req.header("x-user-id") || "portal-user";
const role = (req.header("x-role") || "viewer") as Identity["role"];
req.identity = { orgId, userId, role };
next();
});
app.use((req, res, next) => {
const finish = () => {
if (!req.identity) {
return;
}
recordAudit({
orgId: req.identity.orgId,
actor: req.identity.userId,
action: `${req.method} ${req.path}`,
resource: req.originalUrl,
details: {
statusCode: res.statusCode,
query: req.query,
body: req.body
}
});
};
res.on("finish", finish);
next();
});
function requireRole(allowed: Identity["role"][]) {
return (req: Request, res: Response, next: NextFunction) => {
if (!req.identity || !allowed.includes(req.identity.role)) {
res.status(403).json({ error: "forbidden" });
return;
}
next();
};
}
function buildOrgFilters(query: Request["query"]) {
const clauses: string[] = [];
const params: unknown[] = [];
if (typeof query.name === "string" && query.name.trim()) {
clauses.push("name like ?");
params.push(`%${query.name.trim()}%`);
}
if (typeof query.region === "string" && query.region.trim()) {
clauses.push("region = ?");
params.push(query.region.trim());
}
if (typeof query.tier === "string" && query.tier.trim()) {
clauses.push("tier = ?");
params.push(query.tier.trim());
}
const where = clauses.length ? ` where ${clauses.join(" and ")}` : "";
return { where, params };
}
app.get("/health", (_req, res) => {
const counts = {
organizations: db.prepare("select count(*) as count from organizations").get().count,
transfers: db.prepare("select count(*) as count from transfers").get().count,
auditRecords: db.prepare("select count(*) as count from audit_records").get().count
};
res.json({ ok: true, counts });
});
app.get("/orgs", requireRole(["admin", "manager", "viewer"]), (req, res) => {
const { where, params } = buildOrgFilters(req.query);
const rows = db.prepare(`select * from organizations${where} order by id desc`).all(...params);
res.json({ data: rows });
});
app.post("/orgs", requireRole(["admin"]), (req, res) => {
const { name, orgCode, region, tier } = req.body;
const stmt = db.prepare(
"insert into organizations (name, org_code, region, tier) values (?, ?, ?, ?)"
);
const result = stmt.run(name, orgCode, region, tier);
res.status(201).json({ id: result.lastInsertRowid, name, orgCode, region, tier });
});
app.get("/teams", requireRole(["admin", "manager"]), (req, res) => {
const filters: string[] = [];
const params: unknown[] = [];
if (typeof req.query.orgId === "string" && req.query.orgId.trim()) {
filters.push("org_id = ?");
params.push(Number(req.query.orgId));
}
if (typeof req.query.query === "string" && req.query.query.trim()) {
filters.push("(name like ? or cost_center like ?)");
params.push(`%${req.query.query.trim()}%`, `%${req.query.query.trim()}%`);
}
const where = filters.length ? ` where ${filters.join(" and ")}` : "";
const rows = db.prepare(`select * from teams${where} order by created_at desc`).all(...params);
res.json({ data: rows });
});
app.post("/users", requireRole(["admin", "manager"]), (req, res) => {
const { orgId, teamId, email, displayName, role, status } = req.body;
const result = db
.prepare(
"insert into users (org_id, team_id, email, display_name, role, status) values (?, ?, ?, ?, ?, ?)"
)
.run(orgId, teamId ?? null, email, displayName, role, status || "active");
res.status(201).json({ id: result.lastInsertRowid, orgId, teamId, email, displayName, role });
});
app.post("/beneficiaries", requireRole(["admin", "manager"]), (req, res) => {
const { orgId, name, bankName, routing } = req.body;
const result = db
.prepare("insert into beneficiaries (org_id, name, bank_name, routing) values (?, ?, ?, ?)")
.run(orgId, name, bankName, routing);
res.status(201).json({ id: result.lastInsertRowid });
});
app.post("/transfers", requireRole(["admin", "manager"]), (req, res) => {
const { orgId, accountId, beneficiaryId, amount, currency, memo } = req.body;
const result = db
.prepare(
"insert into transfers (org_id, account_id, beneficiary_id, amount, currency, status, memo, requested_by) values (?, ?, ?, ?, ?, ?, ?, ?)"
)
.run(orgId, accountId, beneficiaryId, amount, currency, "queued", memo || "", req.identity?.userId || "unknown");
res.status(201).json({ id: result.lastInsertRowid, status: "queued" });
});
app.post("/documents/upload", requireRole(["admin", "manager"]), upload.single("file"), (req, res) => {
const file = req.file;
if (!file) {
res.status(400).json({ error: "missing file" });
return;
}
const targetName = `${Date.now()}-${file.originalname}`;
fs.copyFileSync(file.path, storagePath(targetName));
const result = db
.prepare(
"insert into documents (org_id, filename, original_name, mime_type) values (?, ?, ?, ?)"
)
.run(req.identity?.orgId || 1, targetName, file.originalname, file.mimetype);
res.status(201).json({ id: result.lastInsertRowid, filename: targetName });
});
app.get("/documents/:name", requireRole(["admin", "manager", "viewer"]), (req, res) => {
const filePath = storagePath(req.params.name);
if (!fs.existsSync(filePath)) {
res.status(404).json({ error: "not found" });
return;
}
res.setHeader("content-type", "application/octet-stream");
res.send(fs.readFileSync(filePath));
});
app.get("/reports/org-summary", requireRole(["admin", "manager"]), (req, res) => {
const orgId = Number(req.query.orgId || req.identity?.orgId || 1);
const org = db.prepare("select * from organizations where id = ?").get(orgId);
const transfers = db
.prepare("select * from transfers where org_id = ? order by created_at desc limit 25")
.all(orgId);
const title = typeof req.query.title === "string" ? req.query.title : "Treasury Summary";
const html = `
<html>
<body>
<h1>${title}</h1>
<h2>${org?.name ?? "Unknown org"}</h2>
<p>Region: ${org?.region ?? "n/a"} Tier: ${org?.tier ?? "n/a"}</p>
<ul>
${transfers
.map(
(transfer) =>
`<li>${transfer.currency} ${transfer.amount} - ${transfer.status} - ${transfer.memo}</li>`
)
.join("")}
</ul>
</body>
</html>
`;
res.type("html").send(html);
});
app.get("/audit", requireRole(["admin"]), (req, res) => {
const orgId = typeof req.query.orgId === "string" ? Number(req.query.orgId) : req.identity?.orgId;
const rows = db
.prepare("select * from audit_records where org_id = ? order by created_at desc limit 100")
.all(orgId);
res.json({ data: rows });
});
app.use((_req, res) => {
res.status(404).json({ error: "not found" });
});
const port = Number(process.env.PORT || 4100);
app.listen(port, () => {
console.log(`portal listening on ${port}`);
});
portal/src/store.js
const fs = require("fs");
const path = require("path");
const Database = require("better-sqlite3");
const baseDir = path.resolve(__dirname, "..", "data");
const storageDir = path.join(baseDir, "documents");
const dbPath = path.join(baseDir, "portal.db");
fs.mkdirSync(storageDir, { recursive: true });
const db = new Database(dbPath);
db.pragma("journal_mode = WAL");
db.exec(`
create table if not exists organizations (
id integer primary key autoincrement,
name text not null,
org_code text not null unique,
region text not null,
tier text not null,
created_at text not null default (datetime('now'))
);
create table if not exists teams (
id integer primary key autoincrement,
org_id integer not null,
name text not null,
cost_center text not null,
status text not null default 'active',
created_at text not null default (datetime('now'))
);
create table if not exists users (
id integer primary key autoincrement,
org_id integer not null,
team_id integer,
email text not null,
display_name text not null,
role text not null,
status text not null default 'active',
created_at text not null default (datetime('now'))
);
create table if not exists accounts (
id integer primary key autoincrement,
org_id integer not null,
account_no text not null,
currency text not null,
balance numeric not null default 0,
created_at text not null default (datetime('now'))
);
create table if not exists beneficiaries (
id integer primary key autoincrement,
org_id integer not null,
name text not null,
bank_name text not null,
routing text not null,
created_at text not null default (datetime('now'))
);
create table if not exists transfers (
id integer primary key autoincrement,
org_id integer not null,
account_id integer not null,
beneficiary_id integer not null,
amount numeric not null,
currency text not null,
status text not null,
memo text not null,
requested_by text not null,
created_at text not null default (datetime('now'))
);
create table if not exists documents (
id integer primary key autoincrement,
org_id integer not null,
filename text not null,
original_name text not null,
mime_type text not null,
created_at text not null default (datetime('now'))
);
create table if not exists audit_records (
id integer primary key autoincrement,
org_id integer,
actor text not null,
action text not null,
resource text not null,
details text not null,
created_at text not null default (datetime('now'))
);
`);
function seed() {
const count = db.prepare("select count(*) as count from organizations").get().count;
if (count > 0) {
return;
}
const insertOrg = db.prepare("insert into organizations (name, org_code, region, tier) values (?, ?, ?, ?)");
const insertTeam = db.prepare("insert into teams (org_id, name, cost_center, status) values (?, ?, ?, ?)");
const insertUser = db.prepare("insert into users (org_id, team_id, email, display_name, role, status) values (?, ?, ?, ?, ?, ?)");
const insertAccount = db.prepare("insert into accounts (org_id, account_no, currency, balance) values (?, ?, ?, ?)");
const insertBeneficiary = db.prepare("insert into beneficiaries (org_id, name, bank_name, routing) values (?, ?, ?, ?)");
const acme = insertOrg.run("Acme Treasury", "acme", "us-east", "enterprise").lastInsertRowid;
const north = insertOrg.run("Northwind Holdings", "northwind", "eu-west", "growth").lastInsertRowid;
const ops = insertTeam.run(acme, "Operations", "100-ops", "active").lastInsertRowid;
const rev = insertTeam.run(acme, "Revenue", "200-rev", "active").lastInsertRowid;
const comp = insertTeam.run(north, "Compliance", "300-comp", "active").lastInsertRowid;
insertUser.run(acme, ops, "treasury.lead@acme.test", "Dana Ledger", "admin", "active");
insertUser.run(acme, rev, "payments.ops@acme.test", "Mika Flow", "manager", "active");
insertUser.run(north, comp, "risk.officer@northwind.test", "Riley Guard", "viewer", "active");
insertAccount.run(acme, "ACCT-10001", "USD", 1250000);
insertAccount.run(acme, "ACCT-10002", "EUR", 860000);
insertAccount.run(north, "ACCT-20001", "USD", 410000);
insertBeneficiary.run(acme, "Vendor One", "Harbor Bank", "110000");
insertBeneficiary.run(acme, "Payroll Service", "Metro Trust", "220000");
insertBeneficiary.run(north, "Regional Supplier", "Union Bank", "330000");
}
seed();
function recordAudit({ orgId = null, actor, action, resource, details }) {
db.prepare(
"insert into audit_records (org_id, actor, action, resource, details) values (?, ?, ?, ?, ?)"
).run(orgId, actor, action, resource, JSON.stringify(details));
}
function storagePath(name) {
return path.join(storageDir, name);
}
module.exports = {
db,
storageDir,
storagePath,
recordAudit
};
portal/tsconfig.json
{
"compilerOptions": {
"target": "ES2020",
"module": "CommonJS",
"moduleResolution": "Node",
"outDir": "dist",
"rootDir": "src",
"strict": true,
"esModuleInterop": true,
"allowJs": true,
"skipLibCheck": true,
"resolveJsonModule": true
},
"include": ["src/**/*"]
}
risk-api/pom.xml
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.treasury</groupId>
<artifactId>risk-api</artifactId>
<version>1.0.0</version>
<properties>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
</properties>
<dependencies>
<dependency>
<groupId>com.h2database</groupId>
<artifactId>h2</artifactId>
<version>2.3.232</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>2.18.3</version>
</dependency>
</dependencies>
</project>
risk-api/src/main/java/com/treasury/risk/RiskApi.java
package com.treasury.risk;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.sun.net.httpserver.Headers;
import com.sun.net.httpserver.HttpExchange;
import com.sun.net.httpserver.HttpHandler;
import com.sun.net.httpserver.HttpServer;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.URI;
import java.net.URLDecoder;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
public class RiskApi {
private static final ObjectMapper MAPPER = new ObjectMapper();
private static final Path DATA_DIR = Path.of("data");
private static final Path DB_PATH = DATA_DIR.resolve("risk.mv.db");
private static Connection db;
record Identity(long orgId, String userId, String role) {}
@FunctionalInterface
interface RouteHandler {
void handle(HttpExchange exchange) throws Exception;
}
public static void main(String[] args) throws Exception {
Files.createDirectories(DATA_DIR);
initDb();
int port = Integer.parseInt(System.getenv().getOrDefault("PORT", "4400"));
HttpServer server = HttpServer.create(new InetSocketAddress(port), 0);
server.createContext("/health", wrap(RiskApi::health));
server.createContext("/risk/exposures", wrap(requireRoles(RiskApi::exposures, "admin", "manager", "viewer")));
server.createContext("/risk/reviews", wrap(requireRoles(RiskApi::reviews, "admin", "manager")));
server.createContext("/risk/report", wrap(requireRoles(RiskApi::report, "admin", "manager")));
server.createContext("/audit", wrap(requireRoles(RiskApi::audit, "admin")));
server.start();
System.out.println("risk-api listening on " + port);
}
private static void initDb() throws Exception {
db = DriverManager.getConnection("jdbc:h2:file:" + DATA_DIR.resolve("risk").toAbsolutePath());
try (Statement st = db.createStatement()) {
st.execute("""
create table if not exists organizations (
id bigint auto_increment primary key,
name varchar not null,
org_code varchar not null unique,
region varchar not null,
tier varchar not null
)
""");
st.execute("""
create table if not exists teams (
id bigint auto_increment primary key,
org_id bigint not null,
name varchar not null,
cost_center varchar not null,
status varchar not null
)
""");
st.execute("""
create table if not exists users (
id bigint auto_increment primary key,
org_id bigint not null,
team_id bigint,
email varchar not null,
display_name varchar not null,
role varchar not null
)
""");
st.execute("""
create table if not exists accounts (
id bigint auto_increment primary key,
org_id bigint not null,
account_no varchar not null,
currency varchar not null,
balance decimal not null
)
""");
st.execute("""
create table if not exists beneficiaries (
id bigint auto_increment primary key,
org_id bigint not null,
name varchar not null,
bank_name varchar not null,
routing varchar not null
)
""");
st.execute("""
create table if not exists transfers (
id bigint auto_increment primary key,
org_id bigint not null,
account_id bigint not null,
beneficiary_id bigint not null,
amount decimal not null,
currency varchar not null,
status varchar not null,
memo varchar not null,
requested_by varchar not null
)
""");
st.execute("""
create table if not exists risk_reviews (
id bigint auto_increment primary key,
org_id bigint not null,
subject varchar not null,
score decimal not null,
note varchar not null,
created_at timestamp default current_timestamp
)
""");
st.execute("""
create table if not exists audit_records (
id bigint auto_increment primary key,
org_id bigint,
actor varchar not null,
action varchar not null,
resource varchar not null,
details varchar not null,
created_at timestamp default current_timestamp
)
""");
}
try (Statement st = db.createStatement(); ResultSet rs = st.executeQuery("select count(*) from organizations")) {
rs.next();
if (rs.getInt(1) == 0) {
seed();
}
}
}
private static void seed() throws Exception {
try (PreparedStatement org = db.prepareStatement("insert into organizations (name, org_code, region, tier) values (?, ?, ?, ?)")) {
org.setString(1, "Acme Treasury");
org.setString(2, "acme");
org.setString(3, "us-east");
org.setString(4, "enterprise");
org.executeUpdate();
org.setString(1, "Northwind Holdings");
org.setString(2, "northwind");
org.setString(3, "eu-west");
org.setString(4, "growth");
org.executeUpdate();
}
}
private static HttpHandler wrap(RouteHandler handler) {
return exchange -> {
try {
Identity identity = identity(exchange);
exchange.setAttribute("identity", identity);
handler.handle(exchange);
audit(identity.orgId(), identity.userId(), exchange.getRequestMethod() + " " + exchange.getRequestURI().getPath(), exchange.getRequestURI().toString(), Map.of("status", "completed"));
} catch (Exception e) {
sendJson(exchange, 500, Map.of("error", e.getMessage()));
}
};
}
private static RouteHandler requireRoles(RouteHandler handler, String... roles) {
return exchange -> {
Identity identity = (Identity) exchange.getAttribute("identity");
if (identity == null) {
identity = identity(exchange);
exchange.setAttribute("identity", identity);
}
for (String role : roles) {
if (role.equals(identity.role())) {
handler.handle(exchange);
return;
}
}
sendJson(exchange, 403, Map.of("error", "forbidden"));
};
}
private static Identity identity(HttpExchange exchange) {
Headers headers = exchange.getRequestHeaders();
long orgId = parseLong(headers.getFirst("X-Org-Id"), 1L);
String userId = headers.getFirst("X-User-Id");
if (userId == null || userId.isBlank()) {
userId = "risk-user";
}
String role = headers.getFirst("X-Role");
if (role == null || role.isBlank()) {
role = "viewer";
}
return new Identity(orgId, userId, role);
}
private static void health(HttpExchange exchange) throws Exception {
try (Statement st = db.createStatement(); ResultSet rs = st.executeQuery("select count(*) from risk_reviews")) {
rs.next();
sendJson(exchange, 200, Map.of("ok", true, "riskReviews", rs.getInt(1)));
}
}
private static void exposures(HttpExchange exchange) throws Exception {
Identity identity = (Identity) exchange.getAttribute("identity");
Map<String, String> query = query(exchange.getRequestURI());
List<String> clauses = new ArrayList<>();
List<Object> args = new ArrayList<>();
clauses.add("org_id = ?");
args.add(identity.orgId());
if (query.containsKey("status") && !query.get("status").isBlank()) {
clauses.add("status = ?");
args.add(query.get("status"));
}
if (query.containsKey("currency") && !query.get("currency").isBlank()) {
clauses.add("currency = ?");
args.add(query.get("currency"));
}
if (query.containsKey("memo") && !query.get("memo").isBlank()) {
clauses.add("memo like ?");
args.add("%" + query.get("memo") + "%");
}
String sql = "select id, account_id, beneficiary_id, amount, currency, status, memo from transfers where " + String.join(" and ", clauses) + " order by id desc";
List<Map<String, Object>> rows = query(sql, args);
sendJson(exchange, 200, Map.of("data", rows));
}
private static void reviews(HttpExchange exchange) throws Exception {
Identity identity = (Identity) exchange.getAttribute("identity");
if ("POST".equalsIgnoreCase(exchange.getRequestMethod())) {
Map<String, Object> body = readJson(exchange.getRequestBody());
try (PreparedStatement ps = db.prepareStatement("insert into risk_reviews (org_id, subject, score, note) values (?, ?, ?, ?)")) {
ps.setLong(1, identity.orgId());
ps.setString(2, String.valueOf(body.get("subject")));
ps.setBigDecimal(3, new java.math.BigDecimal(String.valueOf(body.get("score"))));
ps.setString(4, String.valueOf(body.get("note")));
ps.executeUpdate();
}
sendJson(exchange, 201, Map.of("created", true));
return;
}
List<Map<String, Object>> rows = query(
"select id, subject, score, note, created_at from risk_reviews where org_id = ? order by id desc",
List.of(identity.orgId())
);
sendJson(exchange, 200, Map.of("data", rows));
}
private static void report(HttpExchange exchange) throws Exception {
Identity identity = (Identity) exchange.getAttribute("identity");
Map<String, String> query = query(exchange.getRequestURI());
String title = query.getOrDefault("title", "Risk Exposure Report");
String subject = query.getOrDefault("subject", "Treasury Portfolio");
String note = query.getOrDefault("note", "No note supplied");
List<Map<String, Object>> rows = query(
"select id, amount, currency, status from transfers where org_id = ? order by id desc limit 10",
List.of(identity.orgId())
);
StringBuilder html = new StringBuilder();
html.append("<html><body><h1>").append(title).append("</h1>");
html.append("<h2>").append(subject).append("</h2>");
html.append("<p>").append(note).append("</p>");
html.append("<ul>");
for (Map<String, Object> row : rows) {
html.append("<li>").append(row.get("currency")).append(" ").append(row.get("amount")).append(" ").append(row.get("status")).append("</li>");
}
html.append("</ul></body></html>");
sendText(exchange, 200, html.toString(), "text/html");
}
private static void audit(HttpExchange exchange) throws Exception {
Identity identity = (Identity) exchange.getAttribute("identity");
List<Map<String, Object>> rows = query(
"select id, actor, action, resource, details, created_at from audit_records where org_id = ? order by id desc limit 100",
List.of(identity.orgId())
);
sendJson(exchange, 200, Map.of("data", rows));
}
private static void audit(long orgId, String actor, String action, String resource, Map<String, Object> details) throws Exception {
try (PreparedStatement ps = db.prepareStatement("insert into audit_records (org_id, actor, action, resource, details) values (?, ?, ?, ?, ?)")) {
ps.setLong(1, orgId);
ps.setString(2, actor);
ps.setString(3, action);
ps.setString(4, resource);
ps.setString(5, MAPPER.writeValueAsString(details));
ps.executeUpdate();
}
}
private static List<Map<String, Object>> query(String sql, List<Object> args) throws Exception {
try (PreparedStatement ps = db.prepareStatement(sql)) {
for (int i = 0; i < args.size(); i++) {
ps.setObject(i + 1, args.get(i));
}
try (ResultSet rs = ps.executeQuery()) {
List<Map<String, Object>> rows = new ArrayList<>();
while (rs.next()) {
Map<String, Object> row = new LinkedHashMap<>();
int cols = rs.getMetaData().getColumnCount();
for (int i = 1; i <= cols; i++) {
row.put(rs.getMetaData().getColumnLabel(i), rs.getObject(i));
}
rows.add(row);
}
return rows;
}
}
}
private static Map<String, String> query(URI uri) {
Map<String, String> out = new HashMap<>();
String raw = uri.getRawQuery();
if (raw == null || raw.isBlank()) {
return out;
}
for (String part : raw.split("&")) {
int idx = part.indexOf('=');
String key = idx >= 0 ? part.substring(0, idx) : part;
String value = idx >= 0 ? part.substring(idx + 1) : "";
out.put(urlDecode(key), urlDecode(value));
}
return out;
}
private static Map<String, Object> readJson(InputStream inputStream) throws IOException {
return MAPPER.readValue(inputStream, new TypeReference<>() {});
}
private static void sendJson(HttpExchange exchange, int status, Map<String, Object> payload) throws IOException {
byte[] data = MAPPER.writeValueAsBytes(payload);
exchange.getResponseHeaders().set("Content-Type", "application/json");
exchange.sendResponseHeaders(status, data.length);
try (OutputStream os = exchange.getResponseBody()) {
os.write(data);
}
}
private static void sendText(HttpExchange exchange, int status, String text, String contentType) throws IOException {
byte[] data = text.getBytes(StandardCharsets.UTF_8);
exchange.getResponseHeaders().set("Content-Type", contentType);
exchange.sendResponseHeaders(status, data.length);
try (OutputStream os = exchange.getResponseBody()) {
os.write(data);
}
}
private static long parseLong(String value, long fallback) {
try {
return value == null ? fallback : Long.parseLong(value);
} catch (NumberFormatException e) {
return fallback;
}
}
private static String urlDecode(String value) {
return URLDecoder.decode(value, StandardCharsets.UTF_8);
}
}