feat: application submission + approval workflow

Phase 3 of auth-and-beginner-flow plan:

- Application submission with required question validation
- Duplicate application prevention (same form + member)
- Application status machine:
  - Beginner: 납부_대기 → 납부_완료 → 승인
  - Conversion: 심사_대기 → 승인
- Payment confirmation (admin)
- Bulk approval with automatic member status update
  - Beginner form → status "Beginner"
  - Conversion form → status "Regular"
- Application cancellation (before payment only)
- Migration 005: applications + application_answers tables

GraphQL: submitApplication, confirmPayment, approveApplications,
cancelApplication mutations + applications, myApplications queries

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: ImTotem

alembic/versions/005_applications.py

@@ -0,0 +1,37 @@
+"""applications and answers
+
+Revision ID: 005
+Revises: 004
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "005"
+down_revision = "004"
+
+
+def upgrade() -> None:
+    op.create_table(
+        "applications",
+        sa.Column("id", sa.String, primary_key=True),
+        sa.Column("form_id", sa.String, sa.ForeignKey("forms.id")),
+        sa.Column("member_id", sa.String, sa.ForeignKey("members.id")),
+        sa.Column("status", sa.String, nullable=False, server_default="납부_대기"),
+        sa.Column("submitted_at", sa.String),
+        sa.Column("approved_at", sa.String),
+        sa.Column("approved_by", sa.String),
+        sa.Column("updated_at", sa.String),
+    )
+    op.create_table(
+        "application_answers",
+        sa.Column("id", sa.String, primary_key=True),
+        sa.Column("application_id", sa.String, sa.ForeignKey("applications.id", ondelete="CASCADE")),
+        sa.Column("question_id", sa.String, sa.ForeignKey("form_questions.id")),
+        sa.Column("value", sa.Text, nullable=False),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("application_answers")
+    op.drop_table("applications")

src/bcsd_api/apply/__init__.py

@@ -0,0 +1,49 @@
+from sqlalchemy import Connection, insert, select, update
+
+from bcsd_api.repository import BaseRepository
+from bcsd_api.tables import application_answers, applications
+
+
+class PgApplicationRepository(BaseRepository):
+    def __init__(self, conn: Connection):
+        super().__init__(conn, applications)
+
+    def find_by_form(self, form_id: str) -> list[dict]:
+        stmt = select(applications).where(applications.c.form_id == form_id)
+        return [row._asdict() for row in self._conn.execute(stmt)]
+
+    def find_by_member(self, member_id: str) -> list[dict]:
+        stmt = select(applications).where(applications.c.member_id == member_id)
+        return [row._asdict() for row in self._conn.execute(stmt)]
+
+    def find_by_form_member(self, form_id: str, member_id: str) -> dict | None:
+        stmt = select(applications).where(
+            applications.c.form_id == form_id,
+            applications.c.member_id == member_id,
+        )
+        row = self._conn.execute(stmt).first()
+        if not row:
+            return None
+        return row._asdict()
+
+    def create(self, row: dict) -> None:
+        self._conn.execute(insert(applications).values(**row))
+
+    def update_fields(self, app_id: str, updates: dict) -> None:
+        self._conn.execute(
+            update(applications).where(applications.c.id == app_id).values(**updates),
+        )
+
+
+class PgAnswerRepository(BaseRepository):
+    def __init__(self, conn: Connection):
+        super().__init__(conn, application_answers)
+
+    def find_by_application(self, application_id: str) -> list[dict]:
+        stmt = select(application_answers).where(
+            application_answers.c.application_id == application_id,
+        )
+        return [row._asdict() for row in self._conn.execute(stmt)]
+
+    def create(self, row: dict) -> None:
+        self._conn.execute(insert(application_answers).values(**row))

src/bcsd_api/apply/pg_repository.py

@@ -0,0 +1,78 @@
+import strawberry
+from strawberry.types import Info
+
+from bcsd_api.graphql.context import GqlContext, require_user
+from bcsd_api.graphql.convert import from_model
+
+from . import service
+from .schema import AnswerRequest
+from .types import AnswerType, ApplicationType, SubmitInput
+
+
+def _to_app_type(app) -> ApplicationType:
+    data = app.model_dump()
+    data["answers"] = [AnswerType(**a) for a in data["answers"]]
+    return ApplicationType(**data)
+
+
+def resolve_submit(info: Info[GqlContext, None], input: SubmitInput) -> ApplicationType:
+    user = require_user(info.context)
+    ctx = info.context
+    answers = [AnswerRequest(question_id=a.question_id, value=a.value) for a in input.answers]
+    app = service.submit(
+        ctx.app_repo, ctx.ans_repo,
+        ctx.form_repo, ctx.question_repo,
+        input.form_id, answers, user["sub"],
+    )
+    return _to_app_type(app)
+
+
+def resolve_applications(
+    info: Info[GqlContext, None], form_id: str,
+) -> list[ApplicationType]:
+    require_user(info.context)
+    ctx = info.context
+    apps = service.list_applications(ctx.app_repo, ctx.ans_repo, form_id)
+    return [_to_app_type(a) for a in apps]
+
+
+def resolve_application(
+    info: Info[GqlContext, None], id: strawberry.ID,
+) -> ApplicationType:
+    require_user(info.context)
+    ctx = info.context
+    app = service.get_application(ctx.app_repo, ctx.ans_repo, id)
+    return _to_app_type(app)
+
+
+def resolve_my_applications(info: Info[GqlContext, None]) -> list[ApplicationType]:
+    user = require_user(info.context)
+    ctx = info.context
+    apps = service.my_applications(ctx.app_repo, ctx.ans_repo, user["sub"])
+    return [_to_app_type(a) for a in apps]
+
+
+def resolve_confirm_payment(
+    info: Info[GqlContext, None], id: strawberry.ID,
+) -> ApplicationType:
+    user = require_user(info.context)
+    app = service.confirm_payment(info.context.app_repo, id, user["sub"])
+    return _to_app_type(app)
+
+
+def resolve_approve(
+    info: Info[GqlContext, None], ids: list[strawberry.ID],
+) -> list[ApplicationType]:
+    user = require_user(info.context)
+    ctx = info.context
+    apps = service.approve(
+        ctx.app_repo, ctx.member_repo, ctx.form_repo,
+        ids, user["sub"],
+    )
+    return [_to_app_type(a) for a in apps]
+
+
+def resolve_cancel(info: Info[GqlContext, None], id: strawberry.ID) -> bool:
+    user = require_user(info.context)
+    service.cancel(info.context.app_repo, id, user["sub"])
+    return True

src/bcsd_api/apply/resolvers.py

@@ -0,0 +1,29 @@
+from pydantic import BaseModel
+
+
+class AnswerRequest(BaseModel):
+    question_id: str
+    value: str
+
+
+class SubmitRequest(BaseModel):
+    form_id: str
+    answers: list[AnswerRequest]
+
+
+class AnswerResponse(BaseModel):
+    id: str
+    question_id: str
+    value: str
+
+
+class ApplicationResponse(BaseModel):
+    id: str
+    form_id: str
+    member_id: str
+    status: str
+    submitted_at: str | None = None
+    approved_at: str | None = None
+    approved_by: str | None = None
+    updated_at: str | None = None
+    answers: list[AnswerResponse] = []

src/bcsd_api/apply/schema.py

@@ -0,0 +1,156 @@
+from datetime import datetime
+
+from bcsd_api.exception import BadRequest, Conflict, Forbidden, NotFound
+from bcsd_api.form.pg_repository import PgFormRepository, PgQuestionRepository
+from bcsd_api.id_gen import generate_id
+from bcsd_api.member.pg_repository import PgMemberRepository
+from bcsd_api.timezone import KST
+
+from .pg_repository import PgAnswerRepository, PgApplicationRepository
+from .schema import AnswerRequest, AnswerResponse, ApplicationResponse
+
+
+def _now() -> str:
+    return datetime.now(KST).isoformat()
+
+
+def submit(
+    app_repo: PgApplicationRepository,
+    ans_repo: PgAnswerRepository,
+    form_repo: PgFormRepository,
+    q_repo: PgQuestionRepository,
+    req_form_id: str,
+    answers: list[AnswerRequest],
+    member_id: str,
+) -> ApplicationResponse:
+    form = form_repo.find_by_id(req_form_id)
+    if not form:
+        raise NotFound(f"form {req_form_id} not found")
+    _check_duplicate(app_repo, req_form_id, member_id)
+    _validate_required(q_repo, req_form_id, answers)
+    status = "심사_대기" if form["type"] == "conversion" else "납부_대기"
+    return _create_app(app_repo, ans_repo, req_form_id, member_id, status, answers)
+
+
+def _check_duplicate(repo: PgApplicationRepository, form_id: str, member_id: str) -> None:
+    if repo.find_by_form_member(form_id, member_id):
+        raise Conflict("already applied to this form")
+
+
+def _validate_required(
+    q_repo: PgQuestionRepository, form_id: str, answers: list[AnswerRequest],
+) -> None:
+    questions = q_repo.find_by_form(form_id)
+    required_ids = {q["id"] for q in questions if q["required"] == "true"}
+    answered_ids = {a.question_id for a in answers}
+    missing = required_ids - answered_ids
+    if missing:
+        raise BadRequest(f"missing required answers: {missing}")
+
+
+def _create_app(
+    app_repo, ans_repo, form_id, member_id, status, answers,
+) -> ApplicationResponse:
+    now = _now()
+    app_id = generate_id("AP")
+    row = {
+        "id": app_id, "form_id": form_id,
+        "member_id": member_id, "status": status,
+        "submitted_at": now, "updated_at": now,
+    }
+    app_repo.create(row)
+    saved = _save_answers(ans_repo, app_id, answers)
+    return ApplicationResponse(**row, answers=saved)
+
+
+def _save_answers(
+    ans_repo: PgAnswerRepository, app_id: str, answers: list[AnswerRequest],
+) -> list[AnswerResponse]:
+    result = []
+    for a in answers:
+        row = {
+            "id": generate_id("AA"),
+            "application_id": app_id,
+            "question_id": a.question_id,
+            "value": a.value,
+        }
+        ans_repo.create(row)
+        result.append(AnswerResponse(**row))
+    return result
+
+
+def list_applications(
+    app_repo: PgApplicationRepository, ans_repo: PgAnswerRepository, form_id: str,
+) -> list[ApplicationResponse]:
+    rows = app_repo.find_by_form(form_id)
+    return [_with_answers(ans_repo, r) for r in rows]
+
+
+def my_applications(
+    app_repo: PgApplicationRepository, ans_repo: PgAnswerRepository, member_id: str,
+) -> list[ApplicationResponse]:
+    rows = app_repo.find_by_member(member_id)
+    return [_with_answers(ans_repo, r) for r in rows]
+
+
+def get_application(
+    app_repo: PgApplicationRepository, ans_repo: PgAnswerRepository, app_id: str,
+) -> ApplicationResponse:
+    row = app_repo.find_by_id(app_id)
+    if not row:
+        raise NotFound(f"application {app_id} not found")
+    return _with_answers(ans_repo, row)
+
+
+def _with_answers(ans_repo: PgAnswerRepository, row: dict) -> ApplicationResponse:
+    answers = [AnswerResponse(**a) for a in ans_repo.find_by_application(row["id"])]
+    return ApplicationResponse(**row, answers=answers)
+
+
+def confirm_payment(
+    app_repo: PgApplicationRepository, app_id: str, admin_id: str,
+) -> ApplicationResponse:
+    row = app_repo.find_by_id(app_id)
+    if not row:
+        raise NotFound(f"application {app_id} not found")
+    if row["status"] != "납부_대기":
+        raise BadRequest("application is not in payment pending status")
+    app_repo.update_fields(app_id, {"status": "납부_완료", "updated_at": _now()})
+    return ApplicationResponse(**app_repo.find_by_id(app_id))
+
+
+def approve(
+    app_repo: PgApplicationRepository,
+    member_repo: PgMemberRepository,
+    form_repo: PgFormRepository,
+    app_ids: list[str],
+    admin_id: str,
+) -> list[ApplicationResponse]:
+    now = _now()
+    result = []
+    for app_id in app_ids:
+        row = app_repo.find_by_id(app_id)
+        if not row:
+            continue
+        form = form_repo.find_by_id(row["form_id"])
+        new_status = "Regular" if form and form["type"] == "conversion" else "Beginner"
+        app_repo.update_fields(app_id, {
+            "status": "승인", "approved_at": now,
+            "approved_by": admin_id, "updated_at": now,
+        })
+        member_repo.update_status(row["member_id"], new_status)
+        result.append(ApplicationResponse(**app_repo.find_by_id(app_id)))
+    return result
+
+
+def cancel(
+    app_repo: PgApplicationRepository, app_id: str, member_id: str,
+) -> None:
+    row = app_repo.find_by_id(app_id)
+    if not row:
+        raise NotFound(f"application {app_id} not found")
+    if row["member_id"] != member_id:
+        raise Forbidden("cannot cancel another member's application")
+    if row["status"] not in ("납부_대기", "심사_대기"):
+        raise BadRequest("cannot cancel after payment or approval")
+    app_repo.update_fields(app_id, {"status": "취소", "updated_at": _now()})