package service // Multi-record writes, in one transaction each. // // api-contract.md §12.1 lists four flows that the frontend performs as a // sequence of independent HTTP calls, with no transaction and no rollback: // hiring a candidate, assigning workers to a posting, screening a whole list, // and submitting a challenge. A failure halfway through leaves the database // inconsistent — an application marked hired with no staff row, or an // assignment with an application still showing as merely shortlisted. // // This file collapses the first two into one endpoint and one transaction // each, which is what §12.1 says Phase 3 should do. The other two are not here: // `useScreenAllCandidates` is n independent PATCHes that are individually // meaningful (a partial screen is not a corrupt state), and `useSubmitChallenge` // writes evidence the talent user owns, which needs the talent-scoped predicate // and is a different shape of problem. // // WHY THE REPOSITORY IS REUSED RATHER THAN HAND-WRITTEN SQL // // Every write below goes through repo.Repo, built over the transaction rather // than the pool. That is deliberate: the repository is where org_id is forced // from the session, where server-derived columns are filled, where values are // bound and cast to their declared types, and where a constraint violation is // translated into the contract's error codes. Writing raw SQL here would mean // re-deriving all four, and getting one of them subtly wrong. import ( "context" "errors" "fmt" "strings" "time" "github.com/jackc/pgx/v5" "github.com/krow/krow-backend/go-api/internal/authctx" "github.com/krow/krow-backend/go-api/internal/domain" "github.com/krow/krow-backend/go-api/internal/repo" ) // maxAssignmentBatch bounds one assign call. // // `useAssignWorkers` assigns a selection from the talent pool, which is a // human-sized list. The cap exists so a single request cannot hold a // transaction open across thousands of inserts, blocking every other write to // these tables for the duration. const maxAssignmentBatch = 200 // TxBeginner is the part of the pool this file needs. An interface rather than // *pgxpool.Pool so the service stays testable and consistent with repo.Querier. type TxBeginner interface { Begin(ctx context.Context) (pgx.Tx, error) } // WorkflowService owns the multi-record flows. type WorkflowService struct { db TxBeginner // now is injectable so tests can assert on generated dates without // depending on the day they run. now func() time.Time } // NewWorkflows builds the workflow service over a pool. func NewWorkflows(db TxBeginner) *WorkflowService { return &WorkflowService{db: db, now: time.Now} } // WithClock replaces the clock. For tests. func (s *WorkflowService) WithClock(now func() time.Time) *WorkflowService { if now != nil { s.now = now } return s } // inTx runs fn inside a transaction, rolling back on any error. // // The rollback is deferred rather than called on each error path: an early // return, a panic in a callee, and an explicit failure all have to undo the // work, and only a deferred rollback covers the second. Rolling back an // already-committed transaction is a no-op in pgx, so the deferred call is safe // on the success path too. func (s *WorkflowService) inTx(ctx context.Context, fn func(tx pgx.Tx) error) error { tx, err := s.db.Begin(ctx) if err != nil { return fmt.Errorf("begin transaction: %w", err) } defer func() { _ = tx.Rollback(ctx) }() if err := fn(tx); err != nil { return err } return tx.Commit(ctx) } func resourceByPath(path string) (*domain.Resource, error) { res, ok := domain.ResourceByPath[path] if !ok { return nil, domain.Internal(fmt.Errorf("service: resource %q is not registered", path)) } return res, nil } /* ── Hire ───────────────────────────────────────────────────────────────── */ // HireResult is what a completed hire returns: both records the flow touched, // so the caller does not need a follow-up read to render the outcome. type HireResult struct { Application domain.Record `json:"application"` Staff domain.Record `json:"staff"` } // WorkerWithRoleResult is what recording a new worker returns: both records the // flow created, so the caller renders the outcome without a follow-up read. type WorkerWithRoleResult struct { Worker domain.Record `json:"worker"` Role domain.Record `json:"role"` } // CreateWorkerWithRole records a NEW person and their first declared role, atomically. // // The third flow this file exists for, and it has the same shape as the two // above: the frontend would otherwise POST a worker profile, then POST an // employee role against the id it came back with, and a failure between them // leaves a worker nobody meant to create with no role and no way to tell them // apart from a real one. // // WHY THIS IS NOT "FIND THE WORKER, THEN ADD A ROLE" // // It is a creation flow, deliberately. A name is not an identity — an // organization may employ any number of people who share one — so nothing here // searches for an existing worker by name, and the caller cannot supply a // worker id. The identity is the email, which the caller states outright, and // the schema decides whether it is already taken: `worker_profiles` carries // UNIQUE (org_id, email) over a `citext` column, so a repeat is a 409 from the // database rather than a check this code could race against. // // Recording a SECOND role for someone who already exists is a different // operation and stays where it was: POST /api/v1/employee-roles, which takes // the worker's email and does not touch worker_profiles. func (s *WorkflowService) CreateWorkerWithRole(ctx context.Context, ident authctx.Identity, body domain.Record) (*WorkerWithRoleResult, error) { profiles, err := resourceByPath("worker-profiles") if err != nil { return nil, err } roles, err := resourceByPath("employee-roles") if err != nil { return nil, err } // The two fields that identify the person. Both are the caller's to supply: // an email is never derived from a name, and a missing one is asked for // rather than invented. name, _ := body["full_name"].(string) email, _ := body["email"].(string) if strings.TrimSpace(name) == "" || strings.TrimSpace(email) == "" { return nil, domain.Validation( "a new worker needs both a name and an email address", map[string]string{"full_name": "required", "email": "required"}) } // The role's own fields, taken from a nested object so that a key meant for // the role can never land on the worker or the other way about. roleBody, _ := body["role"].(map[string]any) if roleBody == nil { roleBody = domain.Record{} } var out WorkerWithRoleResult err = s.inTx(ctx, func(tx pgx.Tx) error { workerRecord := domain.Record{"full_name": name, "email": email} for _, k := range []string{"phone", "availability", "certifications", "experience_years", "skills"} { if v, ok := body[k]; ok { workerRecord[k] = v } } worker, err := repo.New(profiles, tx).Insert(ctx, ident, workerRecord) if err != nil { // A duplicate identity arrives here as the contract's conflict, // translated by the repository from the unique violation. Returned // as it is: the transaction rolls back, so no role is written for a // worker that was never created. return err } // The role points at the worker this transaction just made, never at // one found by name. org_id and created_by are the repository's, from // the session. roleRecord := domain.Record{} for k, v := range roleBody { roleRecord[k] = v } roleRecord["worker_profile_id"] = worker["id"] roleRecord["worker_email"] = worker["email"] roleRecord["worker_name"] = worker["full_name"] role, err := repo.New(roles, tx).Insert(ctx, ident, roleRecord) if err != nil { return err } out = WorkerWithRoleResult{Worker: worker, Role: role} return nil }) if err != nil { return nil, err } return &out, nil } // Hire moves an application to `hired` and creates the staff record, atomically. // // Replaces the two-call sequence at krowHooks.js:302-303. The failure that // motivated it: the PATCH succeeds, the POST fails, and the candidate is now // hired with no employment record and no way for the UI to notice. // // Fields the caller may supply are the ones a hiring form collects — hire_date, // role, phone, profile_tier, status, reviewer_name. Everything else is carried // across from the application, because it is already the truth about this // person and retyping it is how the two records drift apart. func (s *WorkflowService) Hire(ctx context.Context, ident authctx.Identity, applicationID string, body domain.Record) (*HireResult, error) { if !isUUID(applicationID) { return nil, domain.NotFound("JobApplication", applicationID) } apps, err := resourceByPath("job-applications") if err != nil { return nil, err } staffRes, err := resourceByPath("staff") if err != nil { return nil, err } activity, err := resourceByPath("user-activity") if err != nil { return nil, err } var out HireResult err = s.inTx(ctx, func(tx pgx.Tx) error { appRepo := repo.New(apps, tx) // Read inside the transaction. Reading outside it would leave a window // where two concurrent hires both see a not-yet-hired application. app, err := appRepo.Get(ctx, ident, applicationID) if err != nil { return err } if app == nil { return domain.NotFound("JobApplication", applicationID) } // Hiring twice is a conflict, not an idempotent repeat: the second call // would create a second employment record for one application. 409 // rather than 200 because the caller asked for something that cannot be // done, and silently returning the first hire would hide a double // submission rather than report it. if status, _ := app["status"].(string); status == "hired" { return domain.Conflict("this application has already been hired") } email, _ := app["email"].(string) name, _ := app["applicant_name"].(string) if email == "" || name == "" { return domain.Validation( "this application cannot be hired: it has no applicant name or email", map[string]string{"application": "incomplete"}) } staffRecord := domain.Record{ "application_id": applicationID, "job_posting_id": app["job_posting_id"], "worker_profile_id": app["worker_profile_id"], "name": name, "email": email, "phone": pick(body, "phone", app["phone"]), "role": pick(body, "role", app["job_title"]), "hire_date": pick(body, "hire_date", s.now().UTC().Format("2006-01-02")), "ai_score": app["ai_score"], "profile_tier": pick(body, "profile_tier", "Beginner"), "status": pick(body, "status", "onboarding"), "client_rating": app["client_rating"], "reviewer_name": pick(body, "reviewer_name", ""), } // A null worker_profile_id is legitimate — not every applicant has a // profile — but the column list must not carry an explicit nil for a // NOT NULL column, so drop the keys the application had nothing for. dropNil(staffRecord, "worker_profile_id", "job_posting_id") hired, err := appRepo.Update(ctx, ident, applicationID, domain.Record{"status": "hired"}) if err != nil { return err } if hired == nil { // Unreachable: Get above found it under the same predicate. return domain.NotFound("JobApplication", applicationID) } created, err := repo.New(staffRes, tx).Insert(ctx, ident, staffRecord) if err != nil { return err } // The audit entry is part of the transaction on purpose: a hire that // happened without a log entry, or a log entry for a hire that rolled // back, are both worse than neither. if _, err := repo.New(activity, tx).Insert(ctx, ident, domain.Record{ "event_type": "hire_candidate", "details": fmt.Sprintf("%s was hired", name), "application_id": applicationID, "position_id": app["job_posting_id"], "worker_email": email, }); err != nil { return err } out.Application, out.Staff = hired, created return nil }) if err != nil { return nil, err } return &out, nil } /* ── Assign ─────────────────────────────────────────────────────────────── */ // AssignWorker is one worker in an assign request. // // Three ways of naming the application this placement belongs to, in order of // precedence: an id the caller already has, an `application` payload to find or // file one from, and neither — a worker placed straight from the talent pool, // who has no application and is not given an invented one. type AssignWorker struct { WorkerEmail string `json:"worker_email"` WorkerName string `json:"worker_name"` StartsAt string `json:"starts_at"` EndsAt *string `json:"ends_at"` ApplicationID *string `json:"application_id"` WorkerProfileID *string `json:"worker_profile_id"` MatchScore *int `json:"match_score"` Source string `json:"source"` // Application is the application to attach this worker to when no // ApplicationID is supplied: the same fields POST /job-applications takes. // // It exists because an application is what puts a person in the pipeline // for a role, and every downstream step keys on it — the candidate record // is addressed by it, and an AI interview takes one as its subject. A // worker assigned without one is unreachable: nothing to open, and nobody // to interview. The frontend was creating it in a second, untransacted // request; this carries it into the same transaction as the assignment. Application *domain.Record `json:"application"` } // AssignRequest is the body of POST /job-postings/{id}/assignments. type AssignRequest struct { Workers []AssignWorker `json:"workers"` } // AssignResult reports what the batch created. type AssignResult struct { Assignments []domain.Record `json:"assignments"` Count int `json:"count"` } // Assign places workers on a posting, atomically. // // Replaces the 3n sequential round-trips at krowHooks.js:421/449/466 with one // request and one transaction. All-or-nothing across the whole batch: assigning // six workers and having the fourth fail should not leave three assigned, three // not, and the caller unsure which. // // Per worker the flow is: settle the application (see settleApplication — // patch the one named, find-or-file the one described, or neither), insert the // assignment linked to whatever that produced, and write the audit entry. The // application comes first because the assignment references it. func (s *WorkflowService) Assign(ctx context.Context, ident authctx.Identity, postingID string, req AssignRequest) (*AssignResult, error) { if !isUUID(postingID) { return nil, domain.NotFound("JobPosting", postingID) } if len(req.Workers) == 0 { return nil, domain.Validation("at least one worker is required", map[string]string{"workers": "must not be empty"}) } if len(req.Workers) > maxAssignmentBatch { return nil, domain.Validation( fmt.Sprintf("at most %d workers can be assigned in one request", maxAssignmentBatch), map[string]string{"workers": "too many"}) } // Validate the whole batch before opening a transaction. A malformed // request should never have caused a BEGIN. details := map[string]string{} for i, w := range req.Workers { if w.WorkerEmail == "" { details[fmt.Sprintf("workers[%d].worker_email", i)] = "required" } if w.StartsAt == "" { details[fmt.Sprintf("workers[%d].starts_at", i)] = "required" } if w.ApplicationID != nil && !isUUID(*w.ApplicationID) { details[fmt.Sprintf("workers[%d].application_id", i)] = "must be a uuid" } if w.WorkerProfileID != nil && !isUUID(*w.WorkerProfileID) { details[fmt.Sprintf("workers[%d].worker_profile_id", i)] = "must be a uuid" } } if len(details) > 0 { return nil, domain.Validation("assignment payload is not valid", details) } postings, err := resourceByPath("job-postings") if err != nil { return nil, err } assignments, err := resourceByPath("assignments") if err != nil { return nil, err } apps, err := resourceByPath("job-applications") if err != nil { return nil, err } activity, err := resourceByPath("user-activity") if err != nil { return nil, err } out := &AssignResult{Assignments: []domain.Record{}} err = s.inTx(ctx, func(tx pgx.Tx) error { posting, err := repo.New(postings, tx).Get(ctx, ident, postingID) if err != nil { return err } if posting == nil { return domain.NotFound("JobPosting", postingID) } assignRepo := repo.New(assignments, tx) appRepo := repo.New(apps, tx) // The application is filed through the service rather than the // repository so that a payload the caller sent is validated exactly as // POST /job-applications would validate it — unknown fields rejected, // enums checked, required fields demanded — instead of reaching SQL and // coming back as a constraint violation. appSvc := New(apps, tx) activityRepo := repo.New(activity, tx) for i, w := range req.Workers { // The application is settled BEFORE the assignment is written, // because the assignment carries the reference to it and a row // cannot point at one that does not exist yet. An empty id means // this worker legitimately has no application. applicationID, err := s.settleApplication(ctx, ident, appRepo, appSvc, postingID, w) if err != nil { return annotate(err, i) } record := domain.Record{ "job_posting_id": postingID, "worker_email": w.WorkerEmail, "worker_name": w.WorkerName, "starts_at": w.StartsAt, "status": "active", // `owliver` is the column's own default (000001:511) and what // the frontend sends. Substituting `manual` here made the API // and the schema disagree about what an unspecified source // means, so a row written through this endpoint and a row // written through POST /assignments recorded different origins // for the same action. "source": orDefault(w.Source, "owliver"), } if w.EndsAt != nil { record["ends_at"] = *w.EndsAt } if applicationID != "" { record["application_id"] = applicationID } if w.WorkerProfileID != nil { record["worker_profile_id"] = *w.WorkerProfileID } if w.MatchScore != nil { record["match_score"] = *w.MatchScore } created, err := assignRepo.Insert(ctx, ident, record) if err != nil { return annotate(err, i) } out.Assignments = append(out.Assignments, created) entry := domain.Record{ "event_type": "assign_employee", "details": fmt.Sprintf("%s was assigned", orDefault(w.WorkerName, w.WorkerEmail)), "position_id": postingID, "worker_email": w.WorkerEmail, } if applicationID != "" { entry["application_id"] = applicationID } if _, err := activityRepo.Insert(ctx, ident, entry); err != nil { return annotate(err, i) } } return nil }) if err != nil { return nil, err } out.Count = len(out.Assignments) return out, nil } // settleApplication resolves the application an assignment belongs to and // returns its id, or "" when this worker has none. // // Three cases, and the middle one is why this exists: // // - an id was supplied — move that application to `assigned`. Unchanged // behaviour, and it still wins over any payload, because an id is a // decision the caller has already made. // - an `application` payload was supplied — find the application on this // posting for this email, and move it to `assigned` if there is one or file // it if there is not. (job_posting_id, email) is UNIQUE // (job_applications_posting_email_key, 000001), so there is at most one to // find and the insert cannot produce a second. // - neither — nothing. A worker placed straight from the talent pool has no // application, and inventing one for them would be worse than leaving the // link absent. // // Everything here runs inside the caller's transaction, which is what makes the // find-or-create safe: a concurrent assign of the same person to the same // posting blocks on the unique index and is then reported as a conflict, rather // than racing past the lookup and writing a duplicate. func (s *WorkflowService) settleApplication(ctx context.Context, ident authctx.Identity, appRepo *repo.Repo, appSvc *Service, postingID string, w AssignWorker) (string, error) { if w.ApplicationID != nil { updated, err := appRepo.Update(ctx, ident, *w.ApplicationID, domain.Record{"status": "assigned"}) if err != nil { return "", err } if updated == nil { return "", domain.NotFound("JobApplication", *w.ApplicationID) } return *w.ApplicationID, nil } if w.Application == nil { return "", nil } existing, err := findApplication(ctx, ident, appRepo, postingID, w.WorkerEmail) if err != nil { return "", err } if existing != nil { id, _ := existing["id"].(string) updated, err := appRepo.Update(ctx, ident, id, domain.Record{"status": "assigned"}) if err != nil { return "", err } if updated == nil { return "", domain.NotFound("JobApplication", id) } return id, nil } // The posting and the person come from the assignment, not from the // payload. A body naming a different posting or a different email would // file an application about somebody other than the worker being placed, // and the lookup above would never find it again. record := domain.Record{} for k, v := range *w.Application { record[k] = v } record["job_posting_id"] = postingID record["email"] = w.WorkerEmail if _, ok := record["applicant_name"]; !ok { record["applicant_name"] = w.WorkerName } if _, ok := record["status"]; !ok { record["status"] = "assigned" } created, err := appSvc.Create(ctx, ident, record) if err != nil { return "", err } id, _ := created["id"].(string) return id, nil } // findApplication returns this posting's application for this email, or nil. // // The read goes through the repository so it carries the same organization // scope and ownership predicate every other read does — an application in // another tenant is not found, rather than found and then refused. The email // column is citext, so the comparison is case-insensitive: the same equality // the frontend performed with toLowerCase before it had this endpoint. func findApplication(ctx context.Context, ident authctx.Identity, appRepo *repo.Repo, postingID, email string) (domain.Record, error) { res := appRepo.Resource() postingCol, ok := res.Column("job_posting_id") if !ok { return nil, domain.Internal(errors.New("service: job_applications has no job_posting_id column")) } emailCol, ok := res.Column("email") if !ok { return nil, domain.Internal(errors.New("service: job_applications has no email column")) } page, err := appRepo.List(ctx, ident, domain.ListParams{ Limit: 1, Filters: []domain.Filter{ {Column: postingCol, Values: []string{postingID}}, {Column: emailCol, Values: []string{email}}, }, }) if err != nil { return nil, err } if len(page.Records) == 0 { return nil, nil } return page.Records[0], nil } /* ── Helpers ────────────────────────────────────────────────────────────── */ // pick takes the caller's value for a key when they supplied a usable one, and // the fallback otherwise. A present-but-null key means "use the fallback" // rather than "write null", because these columns are NOT NULL. func pick(body domain.Record, key string, fallback any) any { if body == nil { return fallback } v, ok := body[key] if !ok || v == nil { return fallback } if s, isStr := v.(string); isStr && s == "" { return fallback } return v } // dropNil removes keys whose value is nil, so a NOT NULL column is left to its // default instead of being sent an explicit null. func dropNil(rec domain.Record, keys ...string) { for _, k := range keys { if v, ok := rec[k]; ok && v == nil { delete(rec, k) } } } func orDefault(v, fallback string) string { if v == "" { return fallback } return v } // annotate points a validation error at the batch element that produced it. // Without this, "worker_email must not be null" on a batch of forty says // nothing about which one. func annotate(err error, index int) error { var apiErr *domain.Error if !errors.As(err, &apiErr) || apiErr.Details == nil { return err } moved := make(map[string]string, len(apiErr.Details)) for k, v := range apiErr.Details { moved[fmt.Sprintf("workers[%d].%s", index, k)] = v } return domain.Validation(apiErr.Message, moved) }