"""The starter's intake (HR-02, HR-53): a form reached through the platform's PUBLIC-LINK DOOR, never the staff app.
The door holds the token (only its hash is stored), expires it, audits every use (seen apart from submitted) and
answers every failure with the same 404. It calls the two use cases below as a PUBLIC actor scoped to one intake
— `public.subject_of(caller)` must name this intake, or the answer is 404. The staff app's own callers get 404 on
them too: the form is the starter's.
What the form never does: echo a stored sensitive value. It shows which fields are complete and which are
missing, by NAME. Values land in the Sensitive aggregate; the right-to-work upload becomes a document stored in
the bucket and filed in the Sensitive aggregate. The manager sees "details: 5 of 6 done", never a value.
"""
import base64
import hashlib
from typing import Optional
from pydantic import BaseModel, ConfigDict
from erp_platform import public
from erp_platform.bricks import Refused
from erp_platform.types import Name, Note, text
from brick_hr.application import engine
from brick_hr.application import support as S
from brick_hr.application import use_cases as uc
from brick_hr.domain import run as R
from brick_hr.domain.intake import field_values, parse_intake_id, status as intake_status
from brick_hr.domain.sensitive import Sensitive
HR = uc.HR
MAX_UPLOAD = 10 * 1024 * 1024
class Upload(BaseModel):
"""A file the starter uploads (their right-to-work document)."""
model_config = ConfigDict(extra="forbid")
field: Name
filename: Name
content_type: text(100) = "application/octet-stream"
content_base64: str
def _public_or_404(caller: dict, intake_id: str) -> None:
sub = public.subject_of(caller)
if sub is None or sub[0] != "intake" or sub[1] != intake_id or sub[2] != "SUBMIT":
raise Refused(404, "no such form")
def _form(uow, view, intake_id: str) -> tuple:
i = uow.intakes.get(intake_id)
if i is None:
raise Refused(404, "no such form")
fields = view.config.intake_fields(i.pack, i.worker_type)
s = uow.sensitive.get(i.person_id)
held = s.current(view.today) if s else {}
return i, fields, intake_status(fields, held)
@HR.use_case(kind="query", route="GET /intakes/{intake_id}/form")
def read_intake_form(caller: dict, intake_id: Name) -> dict:
"""The starter's form (through the public-link door only): the fields asked for their worker type and country,
which are complete and which are missing — by name. A stored value is never shown back."""
_public_or_404(caller, intake_id)
with S.rules():
with uc.repository() as uow:
view = S.View(uow, engine.SYSTEM)
i, fields, st = _form(uow, view, intake_id)
return {"intake_id": intake_id, "submitted": i.submitted_at is not None,
"fields": [{"field": f["field"], "label": f["label"], "input": f.get("input", "text"),
"required": f.get("required", True), "complete": f["field"] in st["complete"]}
for f in fields],
"done": st["done"], "total": st["total"], "missing": st["missing"]}
@HR.use_case(kind="command", route="POST /intakes/{intake_id}/submit", idempotent=True)
def submit_intake(caller: dict, intake_id: Name, values: Optional[dict[str, Note]] = None,
files: Optional[list[Upload]] = None) -> dict:
"""The starter submits their details (through the public-link door only): address, emergency contact, bank
details (employee) or company and VAT details (freelancer), NI number, and the right-to-work upload. Values go to
the Sensitive aggregate; the answer says which fields are still missing and never repeats a value. The same
submission twice changes nothing."""
_public_or_404(caller, intake_id)
at = S.now()
person_id, n = parse_intake_id(intake_id)
with S.rules():
with uc.repository() as uow:
view = S.View(uow, engine.SYSTEM, at)
i, fields, _st = _form(uow, view, intake_id)
if i.submitted_at is not None:
return {"intake_id": intake_id, "complete": [f["field"] for f in fields], "missing": [],
"submitted": True}
by_name = {f["field"]: f for f in fields}
unknown = sorted(set(values or {}) - {f["field"] for f in fields if f.get("input", "text") == "text"})
if unknown:
raise Refused(422, f"this form does not ask for {', '.join(unknown)}")
s = uow.sensitive.get(person_id) or Sensitive(person_id)
current = s.current(view.today)
for kind, part in field_values(fields, values or {}).items():
merged = {**((current.get(kind) or {}).get("value") or {}), **part}
if merged != (current.get(kind) or {}).get("value"):
s.record(kind, merged, view.today, now=at, by=f"starter:{intake_id}")
stored = []
for up in files or []:
f = by_name.get(up.field)
if f is None or f.get("input") != "file":
raise Refused(422, f"this form has no upload {up.field!r}")
try:
content = base64.b64decode(up.content_base64, validate=True)
except ValueError:
raise Refused(422, f"{up.field}: the file is not base64") from None
if len(content) > MAX_UPLOAD:
raise Refused(413, "that upload is too large")
sha = hashlib.sha256(content).hexdigest()
key = f"hr/{person_id}/sensitive/{f['sensitive_kind']}-{sha[:16]}"
stored.append((key, content, up.content_type))
s.record(f["sensitive_kind"], {"filename": up.filename, "content_type": up.content_type,
"size_bytes": len(content)}, view.today, now=at,
by=f"starter:{intake_id}", object_key=key, sha256=sha)
for key, content, ctype in stored:
uow.objects.put(key, content, ctype)
uow.sensitive.save(s)
# a second unit of work: the Intake aggregate, read back after the Sensitive one committed
with S.rules():
with uc.repository() as uow:
view = S.View(uow, engine.SYSTEM, at)
i, fields, st = _form(uow, view, intake_id)
i.missing_fields = st["missing"]
if not st["missing"]:
i.submitted_at = at
uow.intakes.save(i)
if not st["missing"]:
complete_intake_step(person_id, n, at)
return {"intake_id": intake_id, "complete": st["complete"], "missing": st["missing"],
"submitted": not st["missing"]}
def complete_intake_step(person_id: int, n: int, at) -> None:
def go():
with uc.repository() as uow:
for rid in uow.runs.ids(person_id):
run = uow.runs.get(rid)
if run.n != n or run.cancelled_at is not None:
continue
changed = False
for s in run.ordered():
if s.kind == "intake" and not s.finished:
s.state, s.done_by, s.done_at, s.next_check_at = R.DONE, "starter", at, None
s.last_status, s.manual = "done", False
run.add_log("completed", "starter", at, s.step_no, "details submitted")
changed = True
if changed:
uow.runs.save(run)
engine._committed(go)
@HR.use_case(kind="command", route="POST /intakes/{intake_id}/link", permission=S.PERSON_WRITE)
def reissue_intake_link(caller: dict, intake_id: Name, idempotency_key: Optional[str] = None) -> dict:
"""Copy a fresh link to the starter's form (HR-02): a new link is issued to their personal email and the
previous one stops working. The url is in THIS answer only — it is never stored or logged — so send it now.
(idempotency_key is not honoured here on purpose: replaying this answer would mean storing the link.)"""
at = S.now()
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller, at)
person_id, n = parse_intake_id(intake_id)
p = view.person(person_id)
i = uow.intakes.get(intake_id)
if i is None:
e = p.engagements.get(n)
if e is None:
raise Refused(404, f"no intake {intake_id}")
worker_type, pack = e.worker_type, view.config.setting("pack")
email = (p.detail_at(view.today).data.get("personal_email") or "") if p.detail_at(view.today) else ""
if not email:
raise Refused(409, "the starter has no personal email to send their link to")
cfg = view.config
if i is None:
engine.ensure_intake(person_id, n, worker_type, pack, at)
link = engine.issue_link(intake_id, email, cfg)
return {"intake_id": intake_id, "recipient": email, "url": link.url,
"note": "the previous link no longer works; this url is shown once"}
# ── the public door's handler ──────────────────────────────────────────────────────────────
def _page(form: dict, notice: str = "") -> public.Page:
rows = []
for f in form["fields"]:
state = "done" if f["complete"] else ("required" if f["required"] else "optional")
name = public.esc(f["field"])
if f["input"] == "file":
box = f'<input type="file" name="{name}">'
else:
box = f'<input type="text" name="{name}" autocomplete="off" value="">' # never a stored value
rows.append(f'<p><label>{public.esc(f["label"])} <small>({state})</small><br>{box}</label></p>')
head = f"<p>{public.esc(notice)}</p>" if notice else ""
done = f"<p>{form['done']} of {form['total']} done.</p>"
body = (head + done + '<form method="post" enctype="multipart/form-data">' + "".join(rows)
+ '<p><button type="submit">Save my details</button></p></form>')
return public.Page("Your details", body)
@HR.public("SUBMIT", subject_kind="intake")
def intake_page(link, req) -> public.Page:
"""The starter's form, as the starter sees it: field names, never a stored value. Once complete the link is
spent: it opens the same dead-link page as any other (no oracle)."""
form = read_intake_form(link.actor, link.subject_id)
if form["submitted"] or link.used_at is not None:
raise Refused(404, "no such form")
return _page(form)
@intake_page.act(single_use=False)
def intake_submit(link, req) -> public.Page:
"""Saved in parts until complete; the link is USED (single use) only when nothing is missing."""
values = {k: v for k, v in (req.form or {}).items() if v}
files = [Upload(field=k, filename=f[0] or k, content_type=f[1] or "application/octet-stream",
content_base64=base64.b64encode(f[2]).decode())
for k, f in (req.files or {}).items() if f and f[2]]
out = submit_intake(link.actor, link.subject_id, values=values, files=files)
if out["submitted"]:
link.use()
return public.Page("Thank you", "<p>Thank you — your details are complete. You can close this page.</p>")
return _page(read_intake_form(link.actor, link.subject_id),
f"Saved. Still missing: {', '.join(out['missing'])}.")
__all__ = ["read_intake_form", "submit_intake", "reissue_intake_link", "intake_page"]