"""Create several records at once, so they can cite each other without guessing ids. Why this exists is in services/record_refs.py (#4016): a session that needs records to reference one another used to create them one by one and predict the ids of the ones not yet made — and any concurrent create, from any session or any user, took those numbers. Here the whole batch is ONE transaction: insert every record, flush so the sequence assigns the real ids, rewrite each `{{ref:N}}` with the id and title it names, commit. Other sessions keep creating throughout and cannot interfere — the sequence never hands out the same number twice, so another session's insert just takes a different one. The ids a batch receives are therefore NOT guaranteed consecutive, and nothing here needs them to be: the placeholders are filled with whatever ids came back. Forcing consecutive ids would take a table lock that stalls every user's writes; nobody needs that. All or nothing: a bad placeholder, an invalid status, or a failure mid-insert leaves no partial batch behind — which is also why there are no "reserved" stub records to clean up after a session that dies. """ from __future__ import annotations from dataclasses import dataclass, field from sqlalchemy import select from scribe.models import async_session from scribe.models.milestone import Milestone from scribe.models.note import Note from scribe.services import access as access_svc from scribe.services import notes as notes_svc from scribe.services import systems as systems_svc from scribe.services.record_refs import placeholder_keys, resolve_placeholders # A batch is a plan's steps or a handful of linked records, and every item runs # the near-duplicate gate (an embedding search) before the transaction. 50 is # far past any real plan and keeps one call from becoming a bulk import, which # is a different job with a different door. MAX_BATCH = 50 @dataclass class BatchItem: """One record in a batch. `is_task` False makes it a plain note.""" title: str body: str = "" is_task: bool = True status: str = "todo" priority: str | None = None task_kind: str = "work" tags: list[str] = field(default_factory=list) system_ids: list[int] = field(default_factory=list) @dataclass class BatchMilestone: """The milestone start_planning creates in the same transaction as its steps.""" title: str body: str description: str | None = None def check_batch(items: list[BatchItem], milestone: BatchMilestone | None = None) -> None: """Refuse a batch that cannot be written in full, before any write. Every placeholder must name something the batch will actually create; otherwise the rewrite has nothing to put there, and the alternatives — leaving `{{ref:9}}` in a stored body, or dropping it — both store a reference that points nowhere. """ if not items and milestone is None: raise ValueError("a batch needs at least one record") if len(items) > MAX_BATCH: raise ValueError(f"a batch holds at most {MAX_BATCH} records; got {len(items)}") for i, item in enumerate(items, start=1): if not (item.title or "").strip(): raise ValueError(f"record {i} has no title") valid = {str(i) for i in range(1, len(items) + 1)} if milestone is not None: valid.add("milestone") texts = [item.body for item in items] + ([milestone.body] if milestone else []) unknown = placeholder_keys(*texts) - valid if unknown: named = ", ".join("{{ref:%s}}" % k for k in sorted(unknown)) raise ValueError( f"{named} names no record in this batch. {{{{ref:N}}}} is the Nth " f"record listed (1 to {len(items)})" + (", and {{ref:milestone}} is the milestone being created." if milestone else ".") ) async def create_batch( user_id: int, items: list[BatchItem], *, project_id: int | None = None, milestone_id: int | None = None, milestone: BatchMilestone | None = None, ) -> tuple[Milestone | None, list[Note]]: """Create `items` (and optionally a new milestone they belong to) atomically. `milestone_id` files the items under an EXISTING milestone the caller owns; `milestone` creates a new one alongside them and files them there. Passing both is refused. Returns (the new milestone or None, the notes in input order). Near-duplicate gating is the door's job, exactly as for a single create. """ if milestone is not None and milestone_id: raise ValueError("pass milestone_id (an existing milestone) or a new milestone, not both") check_batch(items, milestone) if milestone_id: async with async_session() as session: existing = (await session.execute( select(Milestone).where( Milestone.id == milestone_id, Milestone.deleted_at.is_(None), ) )).scalars().first() if existing is None: raise ValueError(f"milestone {milestone_id} not found") if project_id and project_id != existing.project_id: raise ValueError( f"milestone {milestone_id} belongs to project {existing.project_id}, not {project_id}" ) project_id = existing.project_id if milestone is not None and not project_id: raise ValueError("a new milestone needs a project_id") # Share-aware (rule 78): a collaborator with write access to a shared # project may plan in it; a bare owner filter would refuse them. if project_id and not await access_svc.can_write_project(user_id, project_id): raise ValueError(f"project {project_id} not found") async with async_session() as session: # Validate every record before adding any: build_note raises on a bad # status/priority, and raising here writes nothing. notes = [ notes_svc.build_note( user_id, title=item.title, body=item.body, tags=item.tags, project_id=project_id, milestone_id=milestone_id, status=item.status if item.is_task else None, priority=item.priority if item.is_task else None, task_kind=notes_svc.minted_kind(item.task_kind) if item.is_task else "work", ) for item in items ] new_ms = None if milestone is not None: new_ms = Milestone( user_id=user_id, project_id=project_id, title=milestone.title, description=milestone.description, body=milestone.body, status="active", ) session.add(new_ms) await session.flush() for note in notes: note.milestone_id = new_ms.id session.add_all(notes) # The flush is where the sequence assigns ids. Nothing is visible to # any other session until the commit below, and if anything between # here and there raises, the context manager rolls it all back. await session.flush() refs = {str(i): f'#{n.id} "{n.title}"' for i, n in enumerate(notes, start=1)} if new_ms is not None: refs["milestone"] = f'milestone {new_ms.id} "{new_ms.title}"' new_ms.body = resolve_placeholders(new_ms.body, refs) for note in notes: note.body = resolve_placeholders(note.body, refs) await session.commit() for note in notes: await session.refresh(note) if new_ms is not None: await session.refresh(new_ms) # After the commit, as a single create does: embedding and System tags are # enrichment on records that now exist, and a failure in either must not # un-create them. for note, item in zip(notes, items): notes_svc.embed_note(note) if item.system_ids: await systems_svc.set_record_systems(user_id, note.id, item.system_ids) if project_id is not None: await notes_svc._maybe_reactivate_project(project_id) return new_ms, notes