"""/api/projects — the product surface. Thin HTTP over services.projects. Access rules (default-deny middleware already guarantees a session): * list/detail: any effective level on the project; unknown OR invisible ids 404 (never 403 — a 403 confirms the id exists). * mutations: projects.manage on that project ("manager side"), except selection, which is the client's direction-locking act (any access). """ import logging from fastapi import APIRouter, Depends, File, HTTPException, UploadFile from fastapi.responses import FileResponse from sqlalchemy import select from app import accounts, bw_config, schemas from app import bw_accounts as bwa from app.config import get_settings from app.db import get_session_factory from app.identity import optional_user from app.models import ( Comment, Option, OptionVariant, Pin, ProjectDetails, Thread, WalkthroughStep, ) from app.services import attachments as att_svc from app.services import stages as stage_svc from app.services import interaction, mockups, outbound from app.services import projects as svc from app.services.rollup import status_body log = logging.getLogger(__name__) router = APIRouter(prefix="/api/projects", tags=["projects"]) def _user(user: str | None = Depends(optional_user)) -> str: # Middleware already 401s anonymous callers; the assert is for type honesty. assert user is not None return user def _not_found(): return HTTPException(status_code=404, detail={ "error_code": "NO_SUCH_PROJECT", "summary": f"No such {bw_config.INSTANCE_NOUN}."}) def _require_view(user: str, project_id: str) -> None: if svc.side_of(user, project_id) is None or not any( i["id"] == project_id for i in bwa.instances()): raise _not_found() @router.get("/{project_id}/people", response_model=schemas.PeopleOut) def people(project_id: str, q: str = "", user: str = Depends(_user)) -> schemas.PeopleOut: """Who can be @mentioned here: exactly the people the server would notify (the project's audience), so the list never offers a name a mention would silently drop. Anyone on the project may ask.""" _require_view(user, project_id) return schemas.PeopleOut(people=[ schemas.PersonOut(**p) for p in interaction.people(project_id, q[:64])]) def _require_manage(user: str, project_id: str) -> None: _require_view(user, project_id) if svc.side_of(user, project_id) != "manager": raise HTTPException(status_code=403, detail={ "error_code": "FORBIDDEN", "summary": "You don't manage this project."}) def _raise_domain(exc: svc.DomainError): raise HTTPException(status_code=exc.status, detail={"error_code": exc.code, "summary": str(exc)}) @router.get("", response_model=schemas.ProjectList) def list_projects(user: str = Depends(_user)) -> schemas.ProjectList: sf = get_session_factory() out = [] with sf() as db: instances = list(accounts.visible_instances(user)) ids = [i["id"] for i in instances] # Two grouped queries rather than two per project: the index is the one # page whose cost grows with how much work you have. activity = svc.activity_index(db, ids) feedback = svc.open_feedback_index(db, ids) for inst in instances: details = db.get(ProjectDetails, inst["id"]) side = svc.side_of(user, inst["id"]) # Count what THIS side can actually open, so a client is never told # about screens and options that are still being prepared. per_screen = [svc.visible_options(side, s.options) for s in svc.visible_screens( side, svc.screens_of(db, inst["id"]))] visible = [opts for opts in per_screen if opts or side == "manager"] state = svc.state_of(db, inst["id"]) out.append(schemas.ProjectSummary( id=inst["id"], label=inst["label"], state=state, open_feedback=feedback.get(inst["id"], 0), last_activity=activity.get(inst["id"]), needs_you=svc.needs_you(state, side), screen_count=len(visible), option_count=sum(len(opts) for opts in visible), sent_at=details.sent_at if details else None, description=(details.description if details else "") or "")) return schemas.ProjectList(projects=out) @router.post("", response_model=schemas.ProjectCreatedOut) def create_project(body: schemas.CreateProjectIn, user: str = Depends(_user)) -> schemas.ProjectCreatedOut: name = body.name.strip() base = svc.slugify(name) existing = {i["id"] for i in bwa.instances()} iid, n = base, 2 while iid in existing: iid, n = f"{base}-{n}", n + 1 try: bwa.create_instance(user, iid, name, audit=accounts.audit) except bwa.AccountsError as exc: code = getattr(exc, "code", "BAD_INPUT") raise HTTPException( status_code=403 if code == "FORBIDDEN" else 400, detail={"error_code": code, "summary": str(exc)}) from exc sf = get_session_factory() with sf() as db: details = svc.ensure_details(db, iid) details.description = body.description.strip() svc.record(db, iid, user, "created", {"label": name}) db.commit() _report_instances() return schemas.ProjectCreatedOut(id=iid, label=name) @router.get("/{project_id}", response_model=schemas.ProjectDetail) def project_detail(project_id: str, user: str = Depends(_user)) -> schemas.ProjectDetail: _require_view(user, project_id) inst = accounts.instance_obj(project_id) if inst is None: raise _not_found() side = svc.side_of(user, project_id) sf = get_session_factory() with sf() as db: details = db.get(ProjectDetails, project_id) screens = svc.visible_screens(side, svc.screens_of(db, project_id)) # What this side may see. A screen whose options are all still fileless # is nothing for a client to act on, so it does not appear for them. options_by_screen = {s.id: svc.visible_options(side, s.options) for s in screens} if side != "manager": screens = [s for s in screens if options_by_screen[s.id]] option_ids = [o.id for s in screens for o in options_by_screen[s.id]] # Computed against the STORED visit, not against now: marking the board # seen is a separate, explicit call the SPA makes after it has rendered, # so this visit still shows what changed. seen_at = svc.last_seen(db, project_id, user) screen_new, option_state = svc.newness(db, screens, seen_at) open_by_option: dict[int, int] = {} steps_by_option: dict[int, int] = {} threads_by_option: dict[int, Thread] = {} if option_ids: for pin, thread in db.execute( select(Pin, Thread).join(Thread, Pin.thread_id == Thread.id) .where(Pin.option_id.in_(option_ids))).all(): if not thread.resolved: open_by_option[pin.option_id] = open_by_option.get(pin.option_id, 0) + 1 for step in db.scalars(select(WalkthroughStep).where( WalkthroughStep.option_id.in_(option_ids))): steps_by_option[step.option_id] = steps_by_option.get(step.option_id, 0) + 1 for t in db.scalars(select(Thread).where( Thread.subject_type == "option", Thread.subject_id.in_([str(i) for i in option_ids]))): threads_by_option[int(t.subject_id)] = t stages = stage_svc.visible(stage_svc.stages_of(db, project_id), side) stage_rows = [schemas.StageOut(**stage_svc.stage_out(st)) for st in stages] db.commit() # the first read of a project seeds its roadmap return schemas.ProjectDetail( id=project_id, label=inst["label"], description=(details.description if details else "") or "", state=svc.state_of(db, project_id), sent_at=details.sent_at if details else None, screens=[schemas.ScreenOut( id=s.id, title=s.title, slug=s.slug, position=s.position, selected_option_id=s.selected_option_id, selected_variants=s.selected_variants, open_comments=(1 if (st := svc.screen_thread(db, s)) is not None and not st.resolved and svc.thread_is_live(st) else 0), comment_count=(0 if st is None else sum(1 for c in st.comments if c.deleted_at is None)), is_new=screen_new.get(s.id, False), published=s.published, options=[schemas.OptionOut( id=o.id, title=o.title, concept_tag=o.concept_tag, position=o.position, has_files=bool(o.files), file_count=len(o.files), walkthrough_count=steps_by_option.get(o.id, 0), open_pin_count=open_by_option.get(o.id, 0), since_last=option_state.get(o.id, ""), blurb=o.blurb or "", thumbnail_url=(f"/api/options/{o.id}/thumbnail?v={o.thumbnail_name[:8]}" if o.thumbnail_name else None), comment_count=svc.comment_count(threads_by_option.get(o.id)), open_comments=(1 if (ot := threads_by_option.get(o.id)) is not None and not ot.resolved and svc.thread_is_live(ot) else 0), variants=[schemas.VariantOut( id=v.id, axis=v.axis, axis_label=v.axis_label or v.axis, key=v.key, label=v.label, css_class=v.css_class, position=v.position, is_default=v.is_default) for v in o.variants], ) for o in options_by_screen[s.id]]) for s in screens], capabilities=schemas.Capabilities(**svc.capabilities(user, project_id)), last_seen=seen_at, new_count=sum(1 for v in option_state.values() if v) + sum(1 for v in screen_new.values() if v), stages=stage_rows, feedback_completed_at=details.feedback_completed_at if details else None, feedback_completed_by=details.feedback_completed_by if details else None, tour_overrides=(details.tour_overrides or {}) if details else {}) @router.patch("/{project_id}", response_model=schemas.OkOut) def update_project(project_id: str, body: schemas.UpdateProjectIn, user: str = Depends(_user)) -> schemas.OkOut: _require_manage(user, project_id) sf = get_session_factory() with sf() as db: details = svc.ensure_details(db, project_id) if body.description is not None: details.description = body.description.strip() db.commit() return schemas.OkOut() @router.get("/{project_id}/status", response_model=schemas.ProjectStatus) def project_status(project_id: str, user: str = Depends(_user)) -> schemas.ProjectStatus: _require_view(user, project_id) sf = get_session_factory() with sf() as db: snap = svc.snapshot(db, project_id) body = status_body(project_id, snap, seq=0, occurred_at="", actors_waiting=svc.waiting_on(db, project_id)) return schemas.ProjectStatus( state=body["state"], counts=schemas.StatusCounts(**body["counts"]), summary=body["summary"]) @router.post("/{project_id}/send", response_model=schemas.OkOut) def send_project(project_id: str, user: str = Depends(_user)) -> schemas.OkOut: _require_manage(user, project_id) inst = accounts.instance_obj(project_id) sf = get_session_factory() with sf() as db: svc.send_project(db, user, project_id) # Moment one of two: they cannot know concepts are waiting by telepathy, # and the bell only speaks to someone already looking at easel. for who in svc.waiting_or_client_side(db, project_id): outbound.queue( db, recipient=who, kind="presented", project_id=project_id, subject=f"Concepts are ready for you: {inst['label']}", url=f"{_app_url()}/p/{project_id}", body=("Your concepts are ready to look through.\n\n" "Open each one, leave a note anywhere on it, and choose " "the direction you would like us to build from.")) interaction.notify_project_side( db, project_id=project_id, actor=user, side="client", kind="turn", category="review", context_label=inst["label"], body="Concepts are ready for your review.", url=f"/p/{project_id}", source_type="project", source_id=project_id, dedupe_prefix=f"sent:{project_id}:{svc.utcnow().isoformat()}") db.commit() return schemas.OkOut() # ------------------------------------------------------------------ screens @router.post("/{project_id}/screens", response_model=schemas.CreatedOut) def create_screen(project_id: str, body: schemas.CreateScreenIn, user: str = Depends(_user)) -> schemas.CreatedOut: _require_manage(user, project_id) sf = get_session_factory() with sf() as db: screen = svc.add_screen(db, project_id, body.title.strip()) svc.record(db, project_id, user, "screen_added", {"screen_id": screen.id, "title": screen.title}) db.commit() return schemas.CreatedOut(id=screen.id) screen_router = APIRouter(prefix="/api/screens", tags=["projects"]) @screen_router.patch("/{screen_id}", response_model=schemas.OkOut) def update_screen(screen_id: int, body: schemas.UpdateScreenIn, user: str = Depends(_user)) -> schemas.OkOut: sf = get_session_factory() with sf() as db: try: screen, project_id = svc.screen_project(db, screen_id) except svc.DomainError as exc: _raise_domain(exc) _require_manage(user, project_id) if body.title is not None: screen.title = body.title.strip() if body.position is not None: screen.position = body.position if body.published is not None: screen.published = body.published svc.record(db, project_id, user, "screen_published" if body.published else "screen_drafted", {"screen_id": screen.id}) db.commit() return schemas.OkOut() @screen_router.delete("/{screen_id}", response_model=schemas.OkOut) def delete_screen(screen_id: int, user: str = Depends(_user)) -> schemas.OkOut: sf = get_session_factory() with sf() as db: try: screen, project_id = svc.screen_project(db, screen_id) except svc.DomainError as exc: _raise_domain(exc) _require_manage(user, project_id) # A screen cascades to its options, so it inherits the same rule. held = [o for o in screen.options if svc.pin_count(db, o.id)] if held: raise HTTPException(status_code=409, detail={ "error_code": "HAS_FEEDBACK", "summary": f"“{screen.title}” has notes on " f"{len(held)} of its options.", "details": "Delete the notes first if you really mean to remove " "this screen — we don't delete a client's feedback " "for them."}) for option in list(screen.options): svc.delete_option(db, option) db.delete(screen) svc.record(db, project_id, user, "screen_deleted", {"screen_id": screen_id}) db.commit() return schemas.OkOut() @screen_router.post("/{screen_id}/select", response_model=schemas.OkOut) def select_option(screen_id: int, body: schemas.SelectOptionIn, user: str = Depends(_user)) -> schemas.OkOut: """The direction-locking act — deliberately open to the client side.""" sf = get_session_factory() with sf() as db: try: screen, project_id = svc.screen_project(db, screen_id) except svc.DomainError as exc: _raise_domain(exc) _require_view(user, project_id) # Clearing used to be team-only, which left the client able to lock a # direction but not to unlock one — an irreversible act performed by the # person with the least context about what it commits us to. The undo is # theirs now; the team is told either way (below), so an unlock is never # silent. try: svc.select_option(db, user, screen, body.option_id, body.variants) except svc.DomainError as exc: _raise_domain(exc) inst = accounts.instance_obj(project_id) if body.option_id is not None: option = db.get(Option, body.option_id) picked = screen.selected_variants or {} # Name the variants in the notification. "Direction picked: The # Direct Answer" is only half the decision when the option carries # axes, and the half it drops is the one we would get wrong. labels = [v.label for v in option.variants if picked.get(v.axis) == v.key] suffix = f" ({', '.join(labels)})" if labels else "" body_text = f"Direction picked: {option.title}{suffix}" dedupe = f"select:{screen.id}:{body.option_id}" else: # An unlock changes what we should be building. Timestamped rather # than keyed on the screen, so a pick/unpick/pick cycle reports every # leg instead of silently deduping the second unlock away. body_text = "Direction unpicked — still deciding" dedupe = f"unselect:{screen.id}:{svc.utcnow().isoformat()}" interaction.notify_project_side( db, project_id=project_id, actor=user, side="manager", kind="turn", category="direction", context_label=f"{inst['label']} · {screen.title}", body=body_text, url=f"/p/{project_id}", source_type="screen", source_id=str(screen.id), dedupe_prefix=dedupe) db.commit() return schemas.OkOut() @screen_router.post("/{screen_id}/options", response_model=schemas.CreatedOut) def create_option(screen_id: int, body: schemas.CreateOptionIn, user: str = Depends(_user)) -> schemas.CreatedOut: sf = get_session_factory() with sf() as db: try: screen, project_id = svc.screen_project(db, screen_id) except svc.DomainError as exc: _raise_domain(exc) _require_manage(user, project_id) option = svc.add_option(db, screen, body.title.strip(), body.concept_tag.strip()) svc.record(db, project_id, user, "option_added", {"screen_id": screen.id, "option_id": option.id}) db.commit() return schemas.CreatedOut(id=option.id) option_router = APIRouter(prefix="/api/options", tags=["projects"]) @option_router.patch("/{option_id}", response_model=schemas.OkOut) def update_option(option_id: int, body: schemas.UpdateOptionIn, user: str = Depends(_user)) -> schemas.OkOut: sf = get_session_factory() with sf() as db: try: option, project_id = svc.option_project(db, option_id) except svc.DomainError as exc: _raise_domain(exc) _require_manage(user, project_id) if body.title is not None: option.title = body.title.strip() if body.concept_tag is not None: option.concept_tag = body.concept_tag.strip() if body.entry_path is not None: option.entry_path = body.entry_path if body.blurb is not None: option.blurb = body.blurb.strip() db.commit() return schemas.OkOut() @option_router.delete("/{option_id}", response_model=schemas.OkOut) def delete_option(option_id: int, user: str = Depends(_user)) -> schemas.OkOut: sf = get_session_factory() with sf() as db: try: option, project_id = svc.option_project(db, option_id) except svc.DomainError as exc: _raise_domain(exc) _require_manage(user, project_id) # An option carrying a conversation is not ours to remove. The standard # says never hard-delete a conversation; the practical reason is that a # mis-click must not be able to destroy a client's feedback. notes = svc.pin_count(db, option.id) if notes: raise HTTPException(status_code=409, detail={ "error_code": "HAS_FEEDBACK", "summary": f"“{option.title}” has {notes} " f"note{'' if notes == 1 else 's'} on it.", "details": "Delete the notes first if you really mean to remove " "this option — we don't delete a client's feedback " "for them."}) svc.delete_option(db, option) svc.record(db, project_id, user, "option_deleted", {"option_id": option_id}) db.commit() return schemas.OkOut() def _report_instances() -> None: """Republish the instance catalog to central (display-only, best-effort).""" from app.config import get_settings if not get_settings().has_bw_client: return try: import app.bw_auth as bw_auth bw_auth.report_instances(bwa.instances()) except Exception: # noqa: BLE001 - visibility must never break a request pass # ------------------------------------------------------------------ variants def _slug(value: str) -> str: out = "".join(c if c.isalnum() else "-" for c in value.strip().lower()) return "-".join(part for part in out.split("-") if part)[:40] @option_router.post("/{option_id}/variants", response_model=schemas.CreatedOut) def create_variant(option_id: int, body: schemas.CreateVariantIn, user: str = Depends(_user)) -> schemas.CreatedOut: """Add a togglable choice to an option (manager side). Variants are authored, not uploaded: the concept's HTML has to carry the CSS for the class, so the person adding one here is the person who wrote it. """ sf = get_session_factory() with sf() as db: try: option, project_id = svc.option_project(db, option_id) except svc.DomainError as exc: _raise_domain(exc) _require_manage(user, project_id) axis = body.axis.strip().lower() key = body.key.strip().lower() css_class = body.css_class.strip() if not axis or not key: raise HTTPException(status_code=400, detail={ "error_code": "BAD_INPUT", "summary": "A variant needs an axis and a label."}) # An empty class is deliberate and load-bearing: it names the design AS # IT SHIPS. A bundle written as `html:not(.v-brand-hive) { … }` has no # class for its plain side, and inventing a do-nothing one to satisfy a # form would put a lie in the data. Every axis needs this choice, or the # single alternative is permanently on and the base is unreachable. if not css_class and any(v.axis == axis and not v.css_class for v in option.variants): existing = next(v.label for v in option.variants if v.axis == axis and not v.css_class) raise HTTPException(status_code=400, detail={ "error_code": "EXISTS", "summary": f"'{existing}' is already the design as it is drawn.", "details": "The second choice is the DIFFERENT look, so it " "needs the class that switches to it, for example " "v-brand-hive. If this design does not contain a " "second look yet, upload a build that does and the " "toggle can be set up in one press."}) if any(v.axis == axis and v.key == key for v in option.variants): raise HTTPException(status_code=400, detail={ "error_code": "EXISTS", "summary": f"'{key}' already exists on the '{axis}' axis."}) if body.is_default: # "Shown first" is a fact about the AXIS, not about one choice — # two defaults on one axis would make the opening view depend on # row order, which is no answer at all. for other in option.variants: if other.axis == axis: other.is_default = False variant = OptionVariant( option_id=option.id, axis=axis, axis_label=body.axis_label.strip() or axis, key=key, label=body.label.strip() or key, css_class=css_class, position=body.position, is_default=body.is_default) db.add(variant) db.flush() svc.record(db, project_id, user, "variant_added", {"option_id": option.id, "axis": axis, "key": key}) db.commit() return schemas.CreatedOut(id=variant.id) @option_router.get("/{option_id}/variant-classes", response_model=schemas.VariantClassesOut) def variant_classes(option_id: int, user: str = Depends(_user)) -> schemas.VariantClassesOut: """Which classes this option's own design scopes to . The authoring form offers these rather than a blank text box: a mistyped class fails SILENTLY (the frame just does not change), which is the worst way to find out. Manager-only — it describes unpublished authoring detail. """ sf = get_session_factory() with sf() as db: try: option, project_id = svc.option_project(db, option_id) except svc.DomainError as exc: _raise_domain(exc) _require_manage(user, project_id) taken = {v.css_class for v in option.variants} classes = mockups.declared_root_classes(db, option) return schemas.VariantClassesOut( classes=classes, unused=[c for c in classes if c not in taken]) @option_router.post("/{option_id}/variants/setup", response_model=schemas.OkOut) def setup_variants(option_id: int, body: schemas.VariantSetupIn, user: str = Depends(_user)) -> schemas.OkOut: """Create a whole toggle at once: the base look plus one choice per class. Adding a variation by hand meant knowing what an axis is, typing a class exactly, and knowing that the plain side needs a choice of its own with no class or the toggle has nothing to switch back to. That is the shape of the table, not of the job. Here the caller sends the labels and the server builds the rest, in one transaction so a half-made toggle cannot exist. """ sf = get_session_factory() with sf() as db: try: option, project_id = svc.option_project(db, option_id) except svc.DomainError as exc: _raise_domain(exc) _require_manage(user, project_id) axis = body.axis.strip().lower() or "version" if any(v.axis == axis for v in option.variants): raise HTTPException(status_code=400, detail={ "error_code": "EXISTS", "summary": f"This option already has a '{axis}' toggle.", "details": "Remove its choices first, or add to it instead."}) if not body.choices: raise HTTPException(status_code=400, detail={ "error_code": "BAD_INPUT", "summary": "A toggle needs at least one look besides the original."}) # The base comes first and is what the client lands on: the design as # its author drew it, before any class is added. rows = [OptionVariant( option_id=option.id, axis=axis, axis_label=body.axis_label.strip() or axis, key="original", label=body.base_label.strip() or "Original", css_class="", position=0, is_default=True)] seen: set[str] = set() for i, choice in enumerate(body.choices, start=1): css = choice.css_class.strip() if not css or css in seen: continue seen.add(css) key = _slug(choice.label) or _slug(css) or f"look-{i}" rows.append(OptionVariant( option_id=option.id, axis=axis, axis_label=body.axis_label.strip() or axis, key=key, label=choice.label.strip() or css, css_class=css, position=i, is_default=False)) if len(rows) < 2: raise HTTPException(status_code=400, detail={ "error_code": "BAD_INPUT", "summary": "A toggle needs at least one look besides the original."}) for row in rows: db.add(row) svc.record(db, project_id, user, "variants_setup", {"option_id": option.id, "axis": axis, "count": len(rows)}) db.commit() return schemas.OkOut() variant_router = APIRouter(prefix="/api/variants", tags=["projects"]) @variant_router.delete("/{variant_id}", response_model=schemas.OkOut) def delete_variant(variant_id: int, user: str = Depends(_user)) -> schemas.OkOut: sf = get_session_factory() with sf() as db: variant = db.get(OptionVariant, variant_id) if variant is None: raise HTTPException(status_code=404, detail={ "error_code": "NO_SUCH_VARIANT", "summary": "No such variant."}) try: _option, project_id = svc.option_project(db, variant.option_id) except svc.DomainError as exc: _raise_domain(exc) _require_manage(user, project_id) db.delete(variant) db.commit() return schemas.OkOut() # ------------------------------------------------- conversation on a screen @screen_router.get("/{screen_id}/thread", response_model=schemas.ScreenThreadOut) def get_screen_thread(screen_id: int, user: str = Depends(_user)) -> schemas.ScreenThreadOut: """The conversation about a screen as a whole. Deliberately distinct from pins: a pin says "about this bit", and there are many; this says "about this page", and there is one. Feedback that is not about a spot had nowhere to go before — the client either invented a location for it or kept it to themselves. """ sf = get_session_factory() with sf() as db: try: screen, project_id = svc.screen_project(db, screen_id) except svc.DomainError as exc: _raise_domain(exc) _require_view(user, project_id) thread = svc.screen_thread(db, screen) if thread is None: return schemas.ScreenThreadOut() return schemas.ScreenThreadOut( thread_id=thread.id, resolved=thread.resolved, comments=[schemas.CommentOut( id=c.id, body_md="" if c.deleted_at else c.body_md, author_username=c.author_username, author_side=svc.side_of(c.author_username, project_id) or "client", created_at=c.created_at, edited_at=c.edited_at, deleted=c.deleted_at is not None, attachment_ids=[a.id for a in c.attachments], reactions=interaction.reactions_of(c)) for c in thread.comments]) @screen_router.post("/{screen_id}/comments", response_model=schemas.CreatedOut) def comment_on_screen(screen_id: int, body: schemas.AddCommentIn, user: str = Depends(_user)) -> schemas.CreatedOut: """Open or continue the screen's conversation. The thread is created on the first comment rather than up front, so a screen nobody has spoken about carries no empty thread.""" sf = get_session_factory() with sf() as db: try: screen, project_id = svc.screen_project(db, screen_id) except svc.DomainError as exc: _raise_domain(exc) _require_view(user, project_id) inst = accounts.instance_obj(project_id) thread = svc.ensure_screen_thread(db, screen, user) interaction.add_comment( db, thread, user, body.body_md, url=f"/p/{project_id}", context_label=f"{inst['label']} · {screen.title}") svc.record(db, project_id, user, "screen_comment", {"screen_id": screen.id, "thread_id": thread.id}) db.commit() return schemas.CreatedOut(id=thread.id) @router.post("/{project_id}/seen", response_model=schemas.OkOut) def mark_project_seen(project_id: str, user: str = Depends(_user)) -> schemas.OkOut: """Record that this person has now looked. Explicit rather than a side effect of the GET: a GET that quietly moved the marker would clear the "new" flags in the very request that returned them, so a client who glanced at the board would be told nothing had changed. The SPA calls this once it has rendered what changed. """ _require_view(user, project_id) sf = get_session_factory() with sf() as db: svc.mark_seen(db, project_id, user) db.commit() return schemas.OkOut() @router.post("/{project_id}/heartbeat", response_model=schemas.OkOut) def heartbeat(project_id: str, user: str = Depends(_user)) -> schemas.OkOut: """"I am still looking at this." Answers the most common anxiety in this workflow — *has he even opened it?* — and doubles as the ecosystem presence ping, which easel only ever sent once at sign-in. `bw_client.report_presence` is throttled and fire-and-forget, so calling it on every beat is free. """ _require_view(user, project_id) sf = get_session_factory() with sf() as db: svc.touch_active(db, project_id, user) db.commit() try: from app import bw_client bw_client.report_presence(user) except Exception: # noqa: BLE001 - presence must never break a request pass return schemas.OkOut() @router.get("/{project_id}/who", response_model=schemas.PresenceOut) def who_is_looking(project_id: str, user: str = Depends(_user)) -> schemas.PresenceOut: """Manager-side only: it answers a question about the client, and showing a client who else has been looking is nobody's business but ours.""" _require_manage(user, project_id) sf = get_session_factory() with sf() as db: return schemas.PresenceOut( people=[schemas.PresenceRowOut(**row) for row in svc.client_presence(db, project_id)]) @router.get("/{project_id}/summary", response_model=schemas.RoundSummaryOut) def round_summary(project_id: str, user: str = Depends(_user)) -> schemas.RoundSummaryOut: """The round in one read. Open to both sides deliberately, and worded from neither: the team needs "what do I owe", the client needs "did what I said land". It is the same list of facts. """ _require_view(user, project_id) sf = get_session_factory() with sf() as db: return schemas.RoundSummaryOut(**svc.round_summary(db, project_id)) def _app_url() -> str: from app.config import get_settings return (get_settings().bw_app_domain or "").rstrip("/") # ------------------------------------------------------------ attachments _ATT_STATUS = {"TOO_LARGE": 413, "SUBJECT_FULL": 409} @router.post("/{project_id}/attachments", response_model=schemas.AttachmentOut, status_code=201) def upload_attachment(project_id: str, file: UploadFile = File(...), user: str = Depends(_user)) -> schemas.AttachmentOut: """A file for this project's conversations — pasted or attached while writing a note (Interaction Standard addendum). It uploads immediately, the response's `markdown` goes into the composer, and the comment is then a normal comment. Same visibility rule as commenting; storage is per project under the data volume, DB-free; the service holds the hostile- surface rules (allow-list, magic bytes, our filenames, caps). """ _require_view(user, project_id) data = file.file.read(att_svc.MAX_BYTES + 1) try: saved = att_svc.save(get_settings().data_dir / "uploads", project_id, f"/api/projects/{project_id}/attachments", file.filename or "", data) except att_svc.AttachmentError as exc: raise HTTPException(status_code=_ATT_STATUS.get(exc.code, 400), detail={ "error_code": exc.code, "summary": str(exc)}) from exc return schemas.AttachmentOut(**saved) @router.get("/{project_id}/attachments/{name}") def get_attachment(project_id: str, name: str, user: str = Depends(_user)) -> FileResponse: """Serve a stored file. Visibility is re-checked through the owning project on every read; the content-type comes from our allow-list, never the stored bytes; anything outside our uuid.ext shape is simply not found.""" _require_view(user, project_id) try: path, ctype = att_svc.resolve(get_settings().data_dir / "uploads", project_id, name) except att_svc.AttachmentError as exc: raise _not_found() from exc return FileResponse(path, media_type=ctype, headers={ "X-Content-Type-Options": "nosniff", "Content-Disposition": f'inline; filename="{name}"', "Cache-Control": "private, max-age=3600", })