Payroll run

// Approve moves a draft run to approved. A second person must approve, before the cut-off.
func (p *Payroll) Approve(ctx context.Context, run int64, approver string, now time.Time) error {
  err := p.tx(ctx, func(tx pgx.Tx) error {
    var cutoff time.Time
    if err := tx.QueryRow(ctx, "SELECT cutoff FROM payroll_runs WHERE id = $1", run).Scan(&cutoff); err != nil {
      return fmt.Errorf("read run %d: %w", run, err)
    }
    if now.After(cutoff) {
      return fmt.Errorf("%w: approved at %s, cut-off %s", ErrPastCutoff, now.Format(time.RFC3339), cutoff.Format(time.RFC3339))
    }
    if err := transition(ctx, tx, run, Draft, Approved, &approver); err != nil {
      return err
    }
    return event(ctx, tx, run, approver, "approve", "")
  })
  var pgErr *pgconn.PgError
  if errors.As(err, &pgErr) && pgErr.Code == "23514" { // check_violation: approved_by <> prepared_by
    return fmt.Errorf("%w: run %d", ErrSameApprover, run)
  }
  return err
}

// transition is a conditional UPDATE: it changes the state only from the state given.
func transition(ctx context.Context, tx pgx.Tx, run int64, from, to string, approver *string) error {
  tag, err := tx.Exec(ctx, q["transition"], run, from, to, approver)
  if err != nil {
    return fmt.Errorf("run %d %s to %s: %w", run, from, to, err)
  }
  if tag.RowsAffected() != 1 {
    return fmt.Errorf("%w: run %d is not %s", ErrState, run, from)
  }
  return nil
}

-- Move a run from one state to the next. 0 rows: the run was not in the expected state.
UPDATE payroll_runs SET state = $3, approved_by = coalesce($4, approved_by)
WHERE id = $1 AND state = $2;

// Process pays every employee of an approved run, one transaction per employee. The item's
// status is the checkpoint, so a worker that dies loses only the employee it was on, and any
// worker can resume the run. It returns how many employees this call posted.
func (p *Payroll) Process(ctx context.Context, run int64, worker string, f Fault) (int, error) {
  err := p.tx(ctx, func(tx pgx.Tx) error {
    err := transition(ctx, tx, run, Approved, Processing, nil)
    if errors.Is(err, ErrState) {
      return nil // already processing: this call resumes it
    }
    if err != nil {
      return err
    }
    return event(ctx, tx, run, worker, "start", "")
  })
  if err != nil {
    return 0, err
  }
  var state string
  if err := p.DB.QueryRow(ctx, "SELECT state FROM payroll_runs WHERE id = $1", run).Scan(&state); err != nil {
    return 0, fmt.Errorf("read run %d: %w", run, err)
  }
  if state != Processing {
    return 0, fmt.Errorf("%w: process run %d in state %s", ErrState, run, state)
  }
  posted, after := 0, int64(0) // after: the last employee this worker posted
  for {
    emp, err := p.payOne(ctx, run, after, posted+1 == f.CrashAt)
    switch {
    case err != nil:
      return posted, err
    case emp == 0 && after == 0:
      return posted, nil
    case emp == 0:
      after = 0 // sweep once more from the start for any employee skipped while locked
    default:
      posted, after = posted+1, emp
    }
  }
}

// payOne posts the next employee after the given one, in one transaction: the accrual entry,
// the payment instruction and the checkpoint commit together. It returns the employee posted,
// or 0 when none is left.
func (p *Payroll) payOne(ctx context.Context, run, after int64, crash bool) (emp int64, err error) {
  conn, err := p.DB.Acquire(ctx)
  if err != nil {
    return 0, fmt.Errorf("acquire connection: %w", err)
  }
  defer conn.Release()
  tx, err := conn.Begin(ctx)
  if err != nil {
    return 0, fmt.Errorf("begin: %w", err)
  }
  defer func() {
    if rbErr := tx.Rollback(ctx); rbErr != nil && !errors.Is(rbErr, pgx.ErrTxClosed) && !crash {
      err = errors.Join(err, fmt.Errorf("rollback: %w", rbErr))
    }
  }()
  var it item
  err = tx.QueryRow(ctx, q["next_item"], run, after).Scan(&it.employee, &it.gross, &it.preTax, &it.incomeTax,
    &it.socialTax, &it.employerTax, &it.postTax, &it.net, &it.reissueOf, &it.account)
  if errors.Is(err, pgx.ErrNoRows) {
    return 0, nil
  }
  if err != nil {
    return 0, fmt.Errorf("next item of run %d: %w", run, err)
  }
  if it.reissueOf == nil { // a reissue pays a liability that the first run already accrued
    if err := accrue(ctx, tx, run, it); err != nil {
      return 0, err
    }
  }
  if crash {
    // The worker process dies here. Its connection ends, and Postgres rolls the transaction back.
    return 0, errors.Join(ErrCrashed, conn.Conn().Close(ctx))
  }
  key := fmt.Sprintf("run:%d:emp:%d", run, it.employee)
  if _, err := tx.Exec(ctx, q["instruct"], key, run, it.employee, it.net, it.account); err != nil {
    return 0, fmt.Errorf("instruction %s: %w", key, err)
  }
  if _, err := tx.Exec(ctx, q["item_posted"], run, it.employee); err != nil {
    return 0, fmt.Errorf("checkpoint employee %d: %w", it.employee, err)
  }
  if err := tx.Commit(ctx); err != nil {
    return 0, fmt.Errorf("commit employee %d: %w", it.employee, err)
  }
  return it.employee, nil
}

-- The checkpoint: the first employee not yet posted, after the last one this worker posted.
-- SKIP LOCKED lets a second worker take the next employee instead of waiting, never the same one.
SELECT i.employee_id, i.gross, i.pre_tax, i.income_tax, i.social_tax, i.employer_tax, i.post_tax,
  i.net, i.reissue_of, e.bank_account
FROM run_items i
JOIN employees e ON e.company_id = i.company_id AND e.id = i.employee_id
WHERE i.run_id = $1 AND i.status = 'calculated' AND i.employee_id > $2
ORDER BY i.employee_id
LIMIT 1
FOR UPDATE OF i SKIP LOCKED;

INSERT INTO journal_entries (idempotency_key, run_id) VALUES ($1, $2)
ON CONFLICT (idempotency_key) DO NOTHING
RETURNING id;

INSERT INTO payment_instructions (idempotency_key, run_id, employee_id, amount_cents, bank_account)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (idempotency_key) DO NOTHING;

UPDATE run_items SET status = 'posted' WHERE run_id = $1 AND employee_id = $2 AND status = 'calculated';

// Reconcile matches the bank's results with the instructions sent. A paid line settles the
// employee's net pay from cash; a returned line leaves it owed. The run ends paid, or
// partially failed when any payment came back.
func (p *Payroll) Reconcile(ctx context.Context, run int64, results []BankResult) (matches []Match, state string, err error) {
  err = p.tx(ctx, func(tx pgx.Tx) error {
    var keys, statuses, reasons []string
    var amounts []int64
    for _, r := range results {
      keys, amounts, statuses, reasons = append(keys, r.Key), append(amounts, r.Amount), append(statuses, r.Status), append(reasons, r.Reason)
    }
    rows, err := tx.Query(ctx, q["reconcile"], run, keys, amounts, statuses, reasons)
    if err != nil {
      return fmt.Errorf("reconcile run %d: %w", run, err)
    }
    matches, err = pgx.CollectRows(rows, func(r pgx.CollectableRow) (Match, error) {
      var m Match
      err := r.Scan(&m.Key, &m.Employee, &m.Amount, &m.Reason, &m.Result)
      return m, err
    })
    if err != nil {
      return fmt.Errorf("reconcile run %d: %w", run, err)
    }
    for _, m := range matches {
      switch m.Result {
      case "paid":
        if err := postEntry(ctx, tx, "settle:"+m.Key, run, m.Employee,
          []string{"net_pay_payable", "cash"}, []int64{m.Amount, -m.Amount}); err != nil {
          return err
        }
        fallthrough
      case "returned":
        if _, err := tx.Exec(ctx, q["settle_status"], m.Key, m.Result, m.Reason); err != nil {
          return fmt.Errorf("mark %s %s: %w", m.Key, m.Result, err)
        }
      }
    }
    var open, returned int
    if err := tx.QueryRow(ctx, q["open_instructions"], run).Scan(&open, &returned); err != nil {
      return fmt.Errorf("count open instructions: %w", err)
    }
    state = Processing
    if open > 0 {
      return nil // wait for the rest of the results
    }
    state = Paid
    if returned > 0 {
      state = PartiallyFailed
    }
    if err := transition(ctx, tx, run, Processing, state, nil); err != nil {
      return err
    }
    return event(ctx, tx, run, "engine", "reconcile", fmt.Sprintf("%d lines, %d returned", len(matches), returned))
  })
  return matches, state, err
}

-- Match the bank's result file with the instructions sent, both ways.
WITH bank AS (
  SELECT * FROM unnest($2::text[], $3::bigint[], $4::text[], $5::text[]) AS b(key, amount, status, reason)
), sent AS (
  SELECT * FROM payment_instructions WHERE run_id = $1 AND status = 'sent'
)
SELECT coalesce(s.idempotency_key, b.key), coalesce(s.employee_id, 0), coalesce(s.amount_cents, 0),
  coalesce(b.reason, ''),
  CASE WHEN s.idempotency_key IS NULL THEN 'unknown to payroll'
       WHEN b.key IS NULL THEN 'missing at the bank'
       WHEN s.amount_cents <> b.amount THEN 'amounts differ'
       ELSE b.status END
FROM sent s
FULL OUTER JOIN bank b ON b.key = s.idempotency_key
ORDER BY 1;