package lhpp

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"
	"github.com/jackc/pgx/v5/pgtype"

	"github.com/TaruLab/keluargamart-backend/internal/db"
	"github.com/TaruLab/keluargamart-backend/internal/distribution"
	"github.com/TaruLab/keluargamart-backend/pkg/seq"
)

// Group lifecycle errors. Every one of these is a caller mistake, surfaced as
// 422 / CodeBusinessViolation — never a 500.
var (
	ErrGroupNotActive            = errors.New("LHPP ini sudah dibatalkan")
	ErrMergeNeedsTwoGroups       = errors.New("merge butuh minimal dua LHPP")
	ErrMergeTargetNotListed      = errors.New("LHPP tujuan harus termasuk daftar yang digabung")
	ErrMergeDriverMismatch       = errors.New("semua LHPP yang digabung harus punya driver yang sama")
	ErrMergeNoMembers            = errors.New("tidak ada DO yang bisa dipindahkan ke LHPP tujuan")
	ErrSplitNoDOs                = errors.New("pilih minimal satu DO untuk dipisah")
	ErrGroupDriverLocked         = errors.New("LHPP ini sudah berisi beberapa DO; pisahkan dulu sebelum ganti driver")
	ErrConcurrentGrouping        = errors.New("LHPP sedang berubah; coba lagi")
	ErrDOAlreadyInAnotherGroup   = errors.New("DO ini sudah masuk LHPP lain")
	errGroupNoMembersToOperateOn = errors.New("LHPP ini tidak punya DO")
)

// seqKeyLHPP is the document_number_sequences prefix for LHPP sheet numbers.
const seqKeyLHPP = "LHPP"

// mutationAttempts bounds the retry on 40001/40P01 (serialization failure,
// deadlock detected). Deliberately small: a lost update here is a human
// clicking merge twice, not a hot path, and the transaction is short.
//
// SERIALIZABLE is not used and must not be introduced — the codebase rejects
// that tier outright (000027: "pg advisory not needed; FOR UPDATE + partial
// unique suffices"), and every mutation here already takes row locks in a fixed
// global order.
const mutationAttempts = 3

// nextLHPPNumber allocates LHPP-YYYYMM-NNNNNN from document_number_sequences.
//
// The allocation runs on the caller's transaction querier, INSIDE the creating
// transaction. That is the whole point: seq.Next is an atomic upsert, so a
// rollback rewinds the counter instead of burning a number. 000084 is the
// precedent — a seeded number with no sequence row made a whole (prefix, month)
// bucket collide forever, and the counter rewinding on rollback meant every
// retry proposed the same dead number.
//
// The month bucket comes from the group's own lhpp_date, so the number's YYYYMM
// and the sequence key always agree — including for the numbers the 000101
// backfill wrote, whose counters that migration advances to match.
func nextLHPPNumber(ctx context.Context, q db.DBTX, at time.Time) (string, error) {
	n, err := seq.Next(ctx, q, seqKeyLHPP, seq.MonthStart(at))
	if err != nil {
		return "", err
	}
	return fmt.Sprintf("%s-%s-%06d", seqKeyLHPP, at.Format("200601"), n), nil
}

// parseGroupDate accepts the YYYY-MM-DD string the DO rows carry. An
// unparseable date is a data hole, and refusing it is better than allocating a
// number in the wrong month bucket.
func parseGroupDate(s string) (time.Time, error) {
	t, err := time.Parse("2006-01-02", s)
	if err != nil {
		return time.Time{}, fmt.Errorf("lhpp: bad group date %q: %w", s, err)
	}
	return t, nil
}

// nullableID turns the "0 means unauthenticated/internal" convention into the
// NULL the audit columns expect, so an internal call never writes a bogus FK to
// user 0.
func nullableID(v int64) pgtype.Int8 {
	if v == 0 {
		return pgtype.Int8{}
	}
	return pgtype.Int8{Int64: v, Valid: true}
}

// nullableIDFromPtr converts an optional assignee: nil clears the column
// (take-back), a value hands the sheet over.
func nullableIDFromPtr(v *int64) pgtype.Int8 {
	if v == nil {
		return pgtype.Int8{}
	}
	return pgtype.Int8{Int64: *v, Valid: true}
}

func derefID(v *int64) int64 {
	if v == nil {
		return 0
	}
	return *v
}

func uniqueSortedIDs(in []int64) []int64 {
	seen := map[int64]bool{}
	out := make([]int64, 0, len(in))
	for _, v := range in {
		if v == 0 || seen[v] {
			continue
		}
		seen[v] = true
		out = append(out, v)
	}
	sort.Slice(out, func(i, j int) bool { return out[i] < out[j] })
	return out
}

// isSerializationFailure reports the two retryable SQLSTATEs: 40001
// (serialization_failure) and 40P01 (deadlock_detected).
func isSerializationFailure(err error) bool {
	var pgErr *pgconn.PgError
	return errors.As(err, &pgErr) && (pgErr.Code == "40001" || pgErr.Code == "40P01")
}

// isDOAlreadyGrouped reports the load-bearing membership violation by
// CONSTRAINT NAME, not by message: lhpp_group_items_do_unique is the invariant
// "a delivery order is in at most one group, ever", and a losing concurrent
// merge is a 409 the caller can retry, not a 500. Matching the name rather than
// the message keeps this from firing on an unrelated 23505, and matching it
// rather than the constraint-vs-index shape is deliberate: PostgreSQL puts the
// violating index's name in that field either way, so the name is the only part
// of this contract that a schema edit can actually break.
func isDOAlreadyGrouped(err error) bool {
	var pgErr *pgconn.PgError
	return errors.As(err, &pgErr) && pgErr.Code == "23505" && pgErr.ConstraintName == constraintDOUnique
}

// withMutationRetry runs a whole transaction, retrying serialization failures
// and deadlock victims a bounded number of times. Retry exhaustion is a 409, not
// a 500 — the caller did nothing wrong, it just lost a race three times.
func withMutationRetry(ctx context.Context, fn func(ctx context.Context) error) error {
	var err error
	for attempt := 0; attempt < mutationAttempts; attempt++ {
		if attempt > 0 {
			select {
			case <-ctx.Done():
				return ctx.Err()
			case <-time.After(time.Duration(attempt) * 25 * time.Millisecond):
			}
		}
		if err = fn(ctx); err == nil || !isSerializationFailure(err) {
			return err
		}
	}
	return ErrConcurrentGrouping
}

// lockedGroups is the state every membership change needs after taking its locks.
type lockedGroups struct {
	rows    map[int64]db.LHPPGetGroupsForUpdateRow
	members map[int64][]int64 // group id -> its delivery order ids (pre-move)
	doDates map[int64]string  // delivery order id -> do_date
	ids     []int64           // ascending, the order the rows were locked in
}

// lockForMutation takes the two locks every membership change needs, in one
// fixed global order:
//
//  1. the delivery_orders rows behind the groups — the same row CreateDispatch
//     already locks, so a merge and a re-dispatch contend for free;
//  2. the lhpp_groups rows themselves, ascending.
//
// Both sets are locked in ascending id order ALWAYS. That is what makes two
// concurrent merges deadlock-free rather than merely rare: they can only ever
// wait on each other, never form a cycle. The group lock covers every
// participating group, not just the ones an operation can empty, because a merge
// that only ADDS a member to group G would otherwise be able to interleave with
// another operation's read of G's members and silently lose that member.
func (s *Service) lockForMutation(ctx context.Context, q *db.Queries, groupIDs []int64) (*lockedGroups, error) {
	ids := uniqueSortedIDs(groupIDs)
	if len(ids) == 0 {
		return nil, ErrGroupNotFound
	}
	memberDOIDs, err := q.LHPPGroupMemberDOIDs(ctx, ids)
	if err != nil {
		return nil, err
	}
	out := &lockedGroups{
		rows:    map[int64]db.LHPPGetGroupsForUpdateRow{},
		members: map[int64][]int64{},
		doDates: map[int64]string{},
		ids:     ids,
	}
	if len(memberDOIDs) > 0 {
		lockedDOs, err := q.LHPPLockDeliveryOrdersForUpdate(ctx, memberDOIDs)
		if err != nil {
			return nil, err
		}
		if len(lockedDOs) != len(memberDOIDs) {
			return nil, ErrDOTidakDitemukan
		}
		for _, d := range lockedDOs {
			out.doDates[d.ID] = d.DoDate
		}
	}
	rows, err := q.LHPPGetGroupsForUpdate(ctx, ids)
	if err != nil {
		return nil, err
	}
	if len(rows) != len(ids) {
		return nil, ErrGroupNotFound
	}
	for _, r := range rows {
		out.rows[r.ID] = r
	}
	for _, doID := range memberDOIDs {
		gid, err := q.LHPPGroupItemDOIDs(ctx, doID)
		if err != nil {
			return nil, err
		}
		out.members[gid] = append(out.members[gid], doID)
	}
	return out, nil
}

// activeRows returns the locked headers, refusing a voided one: a merge, split
// or driver change against a retired sheet is a caller mistake, not something to
// silently apply.
func (l *lockedGroups) activeRows() ([]db.LHPPGetGroupsForUpdateRow, error) {
	rows := make([]db.LHPPGetGroupsForUpdateRow, 0, len(l.ids))
	for _, id := range l.ids {
		r, ok := l.rows[id]
		if !ok {
			return nil, ErrGroupNotFound
		}
		if r.Status != GroupActive {
			return nil, ErrGroupNotActive
		}
		rows = append(rows, r)
	}
	return rows, nil
}

// mergeOnce is one attempt of Merge; see Merge for the contract.
func (s *Service) mergeOnce(ctx context.Context, ids []int64, target int64, reason string, actorID int64) (*Group, error) {
	tx, err := s.repo.Begin(ctx)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback(ctx)
	q := s.repo.Q(tx)

	locked, err := s.lockForMutation(ctx, q, ids)
	if err != nil {
		return nil, err
	}
	if _, err := locked.activeRows(); err != nil {
		return nil, err
	}
	// One group is one driver. Merging across drivers would silently relabel
	// one half of the sheet with the other's name.
	driverID := locked.rows[target].DriverID
	for _, id := range ids {
		if locked.rows[id].DriverID != driverID {
			return nil, ErrMergeDriverMismatch
		}
	}

	sources := make([]int64, 0, len(ids)-1)
	for _, id := range ids {
		if id != target {
			sources = append(sources, id)
		}
	}
	if _, err := q.LHPPMoveGroupItems(ctx, db.LHPPMoveGroupItemsParams{
		ToGroupID:    target,
		FromGroupIds: sources,
		ChangedBy:    nullableID(actorID),
	}); err != nil {
		if isDOAlreadyGrouped(err) {
			return nil, ErrDOAlreadyInAnotherGroup
		}
		return nil, err
	}
	if moved := countMembers(locked, sources); moved == 0 {
		return nil, ErrMergeNoMembers
	}
	// One history row per DO per operation: the 'merged' row IS the per-DO
	// record of this departure. The emptied group's retirement lives on
	// lhpp_groups (status/voided_at/voided_by) — a second per-DO row would
	// contradict "one row per DO per operation" and say nothing the 'merged'
	// row does not already say.
	for _, id := range sources {
		for _, doID := range locked.members[id] {
			if err := s.writeHistory(ctx, q, historyRow{
				DOID:        doID,
				Action:      ActionMerged,
				FromGroupID: id,
				ToGroupID:   target,
				FromDriver:  driverID,
				ToDriver:    driverID,
				Reason:      reason,
				ChangedBy:   actorID,
			}); err != nil {
				return nil, err
			}
		}
	}
	if err := voidEmptied(ctx, q, sources, actorID); err != nil {
		return nil, err
	}
	if err := tx.Commit(ctx); err != nil {
		return nil, err
	}
	return s.Get(ctx, target)
}

// countMembers totals the members a set of groups held before the move.
func countMembers(locked *lockedGroups, groupIDs []int64) int {
	n := 0
	for _, id := range groupIDs {
		n += len(locked.members[id])
	}
	return n
}

// voidEmptied voids every group in ids that no longer holds a DO. Voiding is
// the empty-group mechanism, not a separate feature: a group with no members is
// not a sheet, and lhpp_group_items.group_id is NO ACTION precisely so a group
// must be voided rather than deleted.
func voidEmptied(ctx context.Context, q *db.Queries, groupIDs []int64, actorID int64) error {
	if len(groupIDs) == 0 {
		return nil
	}
	counts, err := q.LHPPCountGroupItemsInGroups(ctx, groupIDs)
	if err != nil {
		return err
	}
	remaining := map[int64]int64{}
	for _, c := range counts {
		remaining[c.GroupID] = c.DoCount
	}
	var empty []int64
	for _, id := range groupIDs {
		if remaining[id] == 0 {
			empty = append(empty, id)
		}
	}
	if len(empty) == 0 {
		return nil
	}
	_, err = q.LHPPVoidGroups(ctx, db.LHPPVoidGroupsParams{
		GroupIds: empty,
		VoidedBy: nullableID(actorID),
	})
	return err
}

// Merge folds several sheets into one for POST /lhpp/groups/merge.
//
// The target keeps its LHPPNumber — a sheet the collector is already carrying
// must not change identity because another DO was folded into it. Every source
// that ends up empty is voided. Money is untouched: per-item figures come from
// the invoice, so only the bucket count changes.
func (s *Service) Merge(ctx context.Context, in MergeInput, actorID int64) (*Group, error) {
	ids := uniqueSortedIDs(in.GroupIDs)
	if len(ids) < 2 {
		return nil, ErrMergeNeedsTwoGroups
	}
	target := in.TargetGroupID
	if target == 0 {
		target = ids[0]
	}
	found := false
	for _, id := range ids {
		if id == target {
			found = true
			break
		}
	}
	if !found {
		return nil, ErrMergeTargetNotListed
	}

	var out *Group
	err := withMutationRetry(ctx, func(ctx context.Context) error {
		var err error
		out, err = s.mergeOnce(ctx, ids, target, in.Reason, actorID)
		return err
	})
	if err != nil {
		return nil, err
	}
	return out, nil
}

// splitOnce is one attempt of Split; see Split for the contract.
func (s *Service) splitOnce(ctx context.Context, groupID int64, doIDs []int64, reason string, actorID int64) (*Group, error) {
	tx, err := s.repo.Begin(ctx)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback(ctx)
	q := s.repo.Q(tx)

	locked, err := s.lockForMutation(ctx, q, []int64{groupID})
	if err != nil {
		return nil, err
	}
	rows, err := locked.activeRows()
	if err != nil {
		return nil, err
	}
	src := rows[0]
	memberSet := map[int64]bool{}
	for _, id := range locked.members[groupID] {
		memberSet[id] = true
	}
	moving := make([]int64, 0, len(doIDs))
	earliest := ""
	for _, doID := range doIDs {
		if !memberSet[doID] {
			// The DO is not on this sheet — either it was never batched or it
			// already moved. Either way this split would silently do nothing
			// for it.
			return nil, ErrDOTidakDitemukan
		}
		moving = append(moving, doID)
		if d := locked.doDates[doID]; earliest == "" || d < earliest {
			earliest = d
		}
	}
	at, err := parseGroupDate(earliest)
	if err != nil {
		return nil, err
	}
	// The new sheet is dated by its EARLIEST member, matching the read
	// projection (date_from/date_to) and the vehicle fields it inherits.
	number, err := nextLHPPNumber(ctx, tx, at)
	if err != nil {
		return nil, err
	}
	newID, err := q.LHPPInsertGroup(ctx, db.LHPPInsertGroupParams{
		LhppNumber: number,
		DriverID:   src.DriverID,
		LhppDate:   pgDate(at),
		CreatedBy:  nullableID(actorID),
	})
	if err != nil {
		return nil, err
	}
	// An UPDATE, not DELETE+INSERT: lhpp_group_items_do_unique covers
	// delivery_order_id alone, so a second membership row for a DO still on the
	// source sheet is rejected by the very invariant that makes this feature
	// safe. Re-pointing the existing row cannot trip it, and there is no window
	// in which the DO sits on no sheet.
	if _, err := q.LHPPMoveGroupItemsByDO(ctx, db.LHPPMoveGroupItemsByDOParams{
		ToGroupID: newID,
		DoIds:     moving,
		ChangedBy: nullableID(actorID),
	}); err != nil {
		return nil, err
	}
	for _, doID := range moving {
		if err := s.writeHistory(ctx, q, historyRow{
			DOID:        doID,
			Action:      ActionSplit,
			FromGroupID: groupID,
			ToGroupID:   newID,
			FromDriver:  src.DriverID,
			ToDriver:    src.DriverID,
			Reason:      reason,
			ChangedBy:   actorID,
		}); err != nil {
			return nil, err
		}
	}
	// A source left with nothing is voided by the same rule a merge uses.
	if err := voidEmptied(ctx, q, []int64{groupID}, actorID); err != nil {
		return nil, err
	}
	if err := tx.Commit(ctx); err != nil {
		return nil, err
	}
	return s.Get(ctx, newID)
}

// Split pulls DOs out of a sheet into a brand new one for
// POST /lhpp/groups/{id}/split. The new group gets a freshly allocated number
// (inside this transaction) and the source keeps its own, so the operation is
// fully reversible: nothing that was already printed changes identity.
func (s *Service) Split(ctx context.Context, groupID int64, in SplitInput, actorID int64) (*Group, error) {
	doIDs := uniqueSortedIDs(in.DOIDs)
	if len(doIDs) == 0 {
		return nil, ErrSplitNoDOs
	}
	var out *Group
	err := withMutationRetry(ctx, func(ctx context.Context) error {
		var err error
		out, err = s.splitOnce(ctx, groupID, doIDs, in.Reason, actorID)
		return err
	})
	if err != nil {
		return nil, err
	}
	return out, nil
}

// VoidOnce-retiring helper: voidOnce is one attempt of Void.
func (s *Service) voidOnce(ctx context.Context, groupID int64, reason string, actorID int64) error {
	tx, err := s.repo.Begin(ctx)
	if err != nil {
		return err
	}
	defer tx.Rollback(ctx)
	q := s.repo.Q(tx)

	locked, err := s.lockForMutation(ctx, q, []int64{groupID})
	if err != nil {
		return err
	}
	rows, err := locked.activeRows()
	if err != nil {
		return err
	}
	src := rows[0]
	doIDs := locked.members[groupID]
	if len(doIDs) == 0 {
		return errGroupNoMembersToOperateOn
	}
	// Deleting the membership rows is what RELEASES these DOs for re-batching:
	// lhpp_group_items_do_unique only constrains rows that exist. A DO released
	// this way gets a fresh single-DO group on its next dispatch, under a new
	// number.
	if _, err := q.LHPPDeleteGroupItems(ctx, []int64{groupID}); err != nil {
		return err
	}
	if _, err := q.LHPPVoidGroups(ctx, db.LHPPVoidGroupsParams{
		GroupIds: []int64{groupID},
		VoidedBy: nullableID(actorID),
	}); err != nil {
		return err
	}
	for _, doID := range doIDs {
		if err := s.writeHistory(ctx, q, historyRow{
			DOID:        doID,
			Action:      ActionVoided,
			FromGroupID: groupID,
			FromDriver:  src.DriverID,
			Reason:      reason,
			ChangedBy:   actorID,
		}); err != nil {
			return err
		}
	}
	return tx.Commit(ctx)
}

// Void retires a sheet for POST /lhpp/groups/{id}/void, releasing its DOs. Get
// still reaches the voided group by id afterwards, so the retired number stays
// auditable.
func (s *Service) Void(ctx context.Context, groupID int64, in VoidInput, actorID int64) error {
	return withMutationRetry(ctx, func(ctx context.Context) error {
		return s.voidOnce(ctx, groupID, in.Reason, actorID)
	})
}

// changeDriverOnce is one attempt of ChangeDriver.
func (s *Service) changeDriverOnce(ctx context.Context, groupID, driverID int64, reason string, actorID int64) (*Group, error) {
	tx, err := s.repo.Begin(ctx)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback(ctx)
	q := s.repo.Q(tx)

	locked, err := s.lockForMutation(ctx, q, []int64{groupID})
	if err != nil {
		return nil, err
	}
	rows, err := locked.activeRows()
	if err != nil {
		return nil, err
	}
	src := rows[0]
	doIDs := locked.members[groupID]
	// driver_id is immutable once the group has more than one member: the sheet
	// a collector is carrying is addressed by driver, and silently re-pointing
	// a multi-DO sheet would hand the next collection round to the wrong
	// person. This is the same rule CreateDispatch enforces, in the same terms.
	if len(doIDs) > 1 {
		return nil, ErrGroupDriverLocked
	}
	if src.DriverID == driverID {
		return &Group{ID: src.ID, LHPPNumber: src.LhppNumber, DriverID: driverID, GroupStatus: src.Status}, tx.Commit(ctx)
	}
	if _, err := q.LHPPUpdateGroupDriver(ctx, db.LHPPUpdateGroupDriverParams{
		DriverID: driverID,
		ID:       groupID,
	}); err != nil {
		return nil, err
	}
	for _, doID := range doIDs {
		if err := s.writeHistory(ctx, q, historyRow{
			DOID:        doID,
			Action:      ActionDriverChanged,
			FromGroupID: groupID,
			ToGroupID:   groupID,
			FromDriver:  src.DriverID,
			ToDriver:    driverID,
			Reason:      reason,
			ChangedBy:   actorID,
		}); err != nil {
			return nil, err
		}
	}
	if err := tx.Commit(ctx); err != nil {
		return nil, err
	}
	return s.Get(ctx, groupID)
}

// ChangeDriver re-points a single-DO sheet at another driver for
// PATCH /lhpp/groups/{id}. The number never changes. A multi-member group is
// refused with ErrGroupDriverLocked — split it first.
func (s *Service) ChangeDriver(ctx context.Context, groupID int64, in UpdateGroupInput, actorID int64) (*Group, error) {
	if in.DriverID <= 0 {
		return nil, ErrDriverRequired
	}
	var out *Group
	err := withMutationRetry(ctx, func(ctx context.Context) error {
		var err error
		out, err = s.changeDriverOnce(ctx, groupID, in.DriverID, in.Reason, actorID)
		return err
	})
	if err != nil {
		return nil, err
	}
	return out, nil
}

// AssignInput is the POST /lhpp/groups/{id}/assign body: the user taking the
// sheet (nil = take it back) and an optional reason for the audit trail.
type AssignInput struct {
	AssigneeID *int64 `json:"assignee_id"`
	Reason     string `json:"reason"`
}

// Assign hands the sheet to a collector — "admin piutang memberi LHPP kepada
// sales untuk dibawa penagihan" — or takes it back (AssigneeID nil), recording
// one 'assigned' history row per member DO in the same transaction. It only
// transfers CUSTODY: recording money still posts through finance.PayInvoice
// (RecordPayment), and the cash's ride to the office remains the warehouse
// deposit custody chain. Accepting a sheet says nothing about payment.
func (s *Service) Assign(ctx context.Context, groupID int64, in AssignInput, actorID int64) (*Group, error) {
	var out *Group
	err := withMutationRetry(ctx, func(ctx context.Context) error {
		var err error
		out, err = s.assignOnce(ctx, groupID, in, actorID)
		return err
	})
	if err != nil {
		return nil, err
	}
	return out, nil
}

func (s *Service) assignOnce(ctx context.Context, groupID int64, in AssignInput, actorID int64) (*Group, error) {
	// Validate the assignee BEFORE opening the tx: a handover to a
	// deactivated or nonexistent account would orphan the sheet (nobody to
	// receive the setoran). nil is the take-back and skips this.
	if in.AssigneeID != nil {
		if _, err := s.repo.q.LHPPAssigneeExists(ctx, *in.AssigneeID); err != nil {
			return nil, fmt.Errorf("%w: penerima %d bukan user aktif", ErrGroupNotFound, *in.AssigneeID)
		}
	}
	tx, err := s.repo.Begin(ctx)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback(ctx)
	q := s.repo.Q(tx)

	locked, err := s.lockForMutation(ctx, q, []int64{groupID})
	if err != nil {
		return nil, err
	}
	rows, err := locked.activeRows()
	if err != nil {
		return nil, err
	}
	src := rows[0]
	doIDs := locked.members[groupID]

	res, err := q.LHPPAssignGroup(ctx, db.LHPPAssignGroupParams{
		AssignedTo: nullableIDFromPtr(in.AssigneeID),
		ID:         groupID,
	})
	if err != nil {
		return nil, err
	}
	if res == 0 {
		return nil, ErrGroupNotActive
	}
	// One history row per member DO, same grain as every other action. The
	// from/to driver columns carry the USER ids (their FK target), which is
	// what an assignee is.
	for _, doID := range doIDs {
		if err := s.writeHistory(ctx, q, historyRow{
			DOID:        doID,
			Action:      ActionAssigned,
			FromGroupID: groupID,
			ToGroupID:   groupID,
			ToDriver:    derefID(in.AssigneeID),
			Reason:      strings.TrimSpace(in.Reason),
			ChangedBy:   actorID,
		}); err != nil {
			return nil, err
		}
	}
	if err := tx.Commit(ctx); err != nil {
		return nil, err
	}
	_ = src
	return s.Get(ctx, groupID)
}

// --- dispatch seam ------------------------------------------------------------

// Grouper is the distribution.LHPPGrouper implementation: ONE single-DO group
// per dispatched delivery order, never auto-batched by (driver, date). That
// automatic bucketing is the behaviour being removed — it made grouping a side
// effect of the query, so a collector could never see unbatched work and a
// printed sheet could change shape on its own.
//
// It runs inside the caller's (CreateDispatch's) transaction on the same DO row
// that CreateDispatch already locked, so a failed group creation rolls the
// dispatch back with it and a merge contends with a re-dispatch for free.
//
// Idempotency falls out of lhpp_group_items_do_unique: a re-dispatch
// re-invokes the seam, finds the DO already placed, and either leaves it alone
// (same driver) or follows the driver (single-member group). If the placement
// still races, the 23505 is swallowed rather than surfaced.
func (s *Service) Grouper(ctx context.Context, q db.DBTX, driverID, doID int64, doDate string) error {
	if driverID <= 0 {
		// No driver means no collector to bill against; there is no sheet.
		return nil
	}
	queries := s.repo.Q(q)
	at, err := parseGroupDate(doDate)
	if err != nil {
		return err
	}
	existing, err := queries.LHPPGroupItemDOIDs(ctx, doID)
	switch {
	case err == nil:
		return s.followDriverChange(ctx, queries, existing, driverID, doID)
	case errors.Is(err, pgx.ErrNoRows):
		return s.createSingleGroup(ctx, q, queries, driverID, doID, at)
	default:
		return err
	}
}

// followDriverChange applies the re-dispatch guard: a DO already sitting in a
// group keeps that group, and the group's driver may only follow while the group
// holds that one DO. Anything else is refused with ErrDOAlreadyBatched, which
// CreateDispatch turns into 422 and rolls the dispatch back.
func (s *Service) followDriverChange(ctx context.Context, q *db.Queries, groupID, driverID, doID int64) error {
	row, err := q.LHPPGetGroup(ctx, groupID)
	if err != nil {
		return err
	}
	if row.DriverID == driverID {
		return nil
	}
	count, err := q.LHPPCountGroupItems(ctx, groupID)
	if err != nil {
		return err
	}
	if count > 1 {
		return distribution.ErrDOAlreadyBatched
	}
	if _, err := q.LHPPUpdateGroupDriver(ctx, db.LHPPUpdateGroupDriverParams{
		DriverID: driverID,
		ID:       groupID,
	}); err != nil {
		return err
	}
	return s.writeHistory(ctx, q, historyRow{
		DOID:        doID,
		Action:      ActionDriverChanged,
		FromGroupID: groupID,
		ToGroupID:   groupID,
		FromDriver:  row.DriverID,
		ToDriver:    driverID,
		Reason:      "re-dispatch",
	})
}

func (s *Service) createSingleGroup(ctx context.Context, q db.DBTX, queries *db.Queries, driverID, doID int64, at time.Time) error {
	number, err := nextLHPPNumber(ctx, q, at)
	if err != nil {
		return err
	}
	groupID, err := queries.LHPPInsertGroup(ctx, db.LHPPInsertGroupParams{
		LhppNumber: number,
		DriverID:   driverID,
		LhppDate:   pgDate(at),
	})
	if err != nil {
		return err
	}
	if _, err := queries.LHPPInsertGroupItem(ctx, db.LHPPInsertGroupItemParams{
		GroupID:         groupID,
		DeliveryOrderID: doID,
	}); err != nil {
		if isDOAlreadyGrouped(err) {
			// The DO was placed by a concurrent dispatch that won the race.
			// Harmless: the DO is batched exactly once, which is the invariant.
			return nil
		}
		return err
	}
	return s.writeHistory(ctx, queries, historyRow{
		DOID:      doID,
		Action:    ActionCreated,
		ToGroupID: groupID,
		ToDriver:  driverID,
		Reason:    "dispatch",
	})
}

// --- history ------------------------------------------------------------------

type historyRow struct {
	DOID        int64
	Action      string
	FromGroupID int64
	ToGroupID   int64
	FromDriver  int64
	ToDriver    int64
	Reason      string
	ChangedBy   int64
}

func (s *Service) writeHistory(ctx context.Context, q *db.Queries, h historyRow) error {
	_, err := q.LHPPInsertGroupHistory(ctx, db.LHPPInsertGroupHistoryParams{
		DeliveryOrderID: h.DOID,
		Action:          h.Action,
		FromGroupID:     nullableID(h.FromGroupID),
		ToGroupID:       nullableID(h.ToGroupID),
		FromDriverID:    nullableID(h.FromDriver),
		ToDriverID:      nullableID(h.ToDriver),
		Reason:          strings.TrimSpace(h.Reason),
		ChangedBy:       nullableID(h.ChangedBy),
	})
	return err
}

// History returns one group's audit trail, newest first.
func (s *Service) History(ctx context.Context, groupID int64) ([]HistoryEntry, error) {
	rows, err := s.repo.q.LHPPGroupHistory(ctx, groupID)
	if err != nil {
		return nil, err
	}
	out := make([]HistoryEntry, 0, len(rows))
	for _, r := range rows {
		e := HistoryEntry{
			ID:     r.ID,
			Action: r.Action,
			Reason: r.Reason,
		}
		if r.DeliveryOrderID.Valid {
			v := r.DeliveryOrderID.Int64
			e.DOID = &v
		}
		if r.FromGroupID.Valid {
			v := r.FromGroupID.Int64
			e.FromGroupID = &v
		}
		if r.ToGroupID.Valid {
			v := r.ToGroupID.Int64
			e.ToGroupID = &v
		}
		if r.FromDriverID.Valid {
			v := r.FromDriverID.Int64
			e.FromDriver = &v
		}
		if r.ToDriverID.Valid {
			v := r.ToDriverID.Int64
			e.ToDriver = &v
		}
		if r.ChangedBy.Valid {
			v := r.ChangedBy.Int64
			e.ChangedBy = &v
		}
		out = append(out, e)
	}
	return out, nil
}

func pgDate(t time.Time) pgtype.Date {
	return pgtype.Date{Time: t, Valid: true}
}