XQKqueue/server/internal/httpapi/staff.go

package httpapi

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"strings"
	"time"
	"unicode/utf8"

	"calllinesystem/server/internal/domain"
	"calllinesystem/server/internal/model"
	"calllinesystem/server/internal/security"

	"github.com/google/uuid"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

func (s *Server) staffProjects(w http.ResponseWriter, r *http.Request) {
	projects, err := s.projectsForUser(r.Context(), currentPrincipal(r.Context()).User)
	if err != nil {
		writeError(w, err)
		return
	}
	writeJSON(w, http.StatusOK, map[string]any{"projects": projectViews(projects)})
}

func projectViews(projects []model.Project) []map[string]any {
	views := make([]map[string]any, 0, len(projects))
	for _, project := range projects {
		views = append(views, projectView(project))
	}
	return views
}

func projectView(project model.Project) map[string]any {
	return map[string]any{
		"id": project.ID, "code": project.Code, "name": project.Name, "status": project.Status,
		"timezone": project.Timezone, "ticket_prefix": project.TicketPrefix,
		"call_mode":                 project.CallMode,
		"default_call_ticket_count": project.CallBatchSize, "call_batch_size": project.CallBatchSize, "batch_size": project.CallBatchSize,
		"max_call_ticket_count":     project.MaxCallTicketCount,
		"default_call_people_count": project.DefaultCallPeopleCount, "max_call_people_count": project.MaxCallPeopleCount,
		"min_party_size": project.MinPartySize, "max_party_size": project.MaxPartySize,
		"grace_period_minutes":     project.GracePeriodMinutes,
		"experienced_people_start": project.ExperiencedPeopleStart,
		"visitor_notice":           project.VisitorNotice,
		"eta": map[string]any{
			"interval_per_person_seconds": project.ETAIntervalSeconds,
		},
	}
}

func (s *Server) queueSnapshot(w http.ResponseWriter, r *http.Request) {
	projectID := r.PathValue("id")
	if err := validateUUID(projectID); err != nil {
		writeError(w, err)
		return
	}
	if err := s.authorizeProject(r.Context(), projectID); err != nil {
		writeError(w, err)
		return
	}
	var project model.Project
	if err := s.db.WithContext(r.Context()).First(&project, "id = ?", projectID).Error; err != nil {
		writeError(w, mapNotFound(err, "PROJECT_NOT_FOUND", "项目不存在"))
		return
	}
	session, err := s.currentQueueSession(s.db.WithContext(r.Context()), project)
	if errors.Is(err, gorm.ErrRecordNotFound) {
		writeJSON(w, http.StatusOK, map[string]any{
			"project": projectView(project), "session": nil, "revision": 0,
			"counts": map[string]int64{}, "people_counts": map[string]int64{}, "waiting": []any{}, "current_batch": nil, "recent_batches": []any{},
			"metrics": map[string]any{
				"waiting_count": 0, "waiting_ticket_count": 0, "waiting_people_count": 0,
				"called_count": 0, "called_ticket_count": 0, "called_people_count": 0,
				"estimated_wait": nil, "last_ticket_number": nil, "next_ticket_number": nil,
			},
			"server_time": s.now(), "last_success_at": s.now(),
		})
		return
	}
	if err != nil {
		writeError(w, err)
		return
	}

	type statusCount struct {
		Status      string
		TicketCount int64 `gorm:"column:ticket_count"`
		PeopleCount int64 `gorm:"column:people_count"`
	}
	var grouped []statusCount
	if err := s.db.WithContext(r.Context()).Model(&model.QueueTicket{}).
		Select("status, count(*) AS ticket_count, COALESCE(sum(party_size), 0) AS people_count").Where("project_id = ? AND queue_session_id = ?", projectID, session.ID).
		Group("status").Scan(&grouped).Error; err != nil {
		writeError(w, err)
		return
	}
	ticketCounts := make(map[string]int64, len(grouped))
	peopleCounts := make(map[string]int64, len(grouped))
	for _, row := range grouped {
		ticketCounts[row.Status] = row.TicketCount
		peopleCounts[row.Status] = row.PeopleCount
	}
	var waiting []model.QueueTicket
	if err := s.db.WithContext(r.Context()).Where("project_id = ? AND queue_session_id = ? AND status = ?", projectID, session.ID, model.TicketWaiting).
		Order("ticket_number ASC").Limit(200).Find(&waiting).Error; err != nil {
		writeError(w, err)
		return
	}
	waitingViews := make([]map[string]any, 0, len(waiting))
	for _, ticket := range waiting {
		view, err := s.staffTicketView(ticket)
		if err != nil {
			writeError(w, err)
			return
		}
		waitingViews = append(waitingViews, view)
	}

	batchView, err := s.currentBatchView(r.Context(), projectID, session.ID)
	if err != nil {
		writeError(w, err)
		return
	}
	nextTicketNumber, err := domain.DisplayNumber("", session.NextTicketNumber)
	if err != nil {
		writeError(w, err)
		return
	}
	var lastTicketNumber any
	if session.NextTicketNumber > 1 {
		formatted, formatErr := domain.DisplayNumber("", session.NextTicketNumber-1)
		if formatErr != nil {
			writeError(w, formatErr)
			return
		}
		lastTicketNumber = formatted
	}
	var lastWaiting model.QueueTicket
	lastWaitingErr := s.db.WithContext(r.Context()).
		Select("id", "party_size").
		Where("project_id = ? AND queue_session_id = ? AND status = ?", projectID, session.ID, model.TicketWaiting).
		Order("ticket_number DESC").First(&lastWaiting).Error
	if lastWaitingErr != nil && !errors.Is(lastWaitingErr, gorm.ErrRecordNotFound) {
		writeError(w, lastWaitingErr)
		return
	}
	waitingPeople := peopleCounts[model.TicketWaiting]
	peopleAheadOfLast := int(waitingPeople)
	if lastWaitingErr == nil {
		peopleAheadOfLast = max(0, peopleAheadOfLast-lastWaiting.PartySize)
	}
	estimatedWait, err := domain.CalculateETA(domain.ETAInput{
		PeopleAhead:       peopleAheadOfLast,
		IntervalPerPerson: time.Duration(project.ETAIntervalSeconds) * time.Second,
		Running:           project.Status == model.ProjectRunning && session.Status == "RUNNING" && ticketCounts[model.TicketWaiting] > 0,
	})
	if err != nil {
		writeError(w, err)
		return
	}
	writeJSON(w, http.StatusOK, map[string]any{
		"project":  projectView(project),
		"session":  map[string]any{"id": session.ID, "business_date": session.BusinessDate.Format("2006-01-02"), "status": session.Status},
		"revision": session.Revision, "counts": ticketCounts, "people_counts": peopleCounts, "waiting": waitingViews, "current_batch": batchView,
		"metrics": map[string]any{
			"waiting_count":        ticketCounts[model.TicketWaiting],
			"waiting_ticket_count": ticketCounts[model.TicketWaiting],
			"waiting_people_count": peopleCounts[model.TicketWaiting],
			"called_count":         ticketCounts[model.TicketCalled] + ticketCounts[model.TicketArrived],
			"called_ticket_count":  ticketCounts[model.TicketCalled] + ticketCounts[model.TicketArrived],
			"called_people_count":  peopleCounts[model.TicketCalled] + peopleCounts[model.TicketArrived],
			"estimated_wait":       estimatedWait,
			"last_ticket_number":   lastTicketNumber,
			"next_ticket_number":   nextTicketNumber,
		},
		"server_time": s.now(), "last_success_at": s.now(),
	})
}

func (s *Server) currentBatchView(ctx context.Context, projectID, sessionID string) (any, error) {
	var batch model.CallBatch
	err := s.db.WithContext(ctx).Where("project_id = ? AND queue_session_id = ? AND status = 'CALLED'", projectID, sessionID).
		Order("batch_sequence DESC").First(&batch).Error
	if errors.Is(err, gorm.ErrRecordNotFound) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	tickets, err := batchTickets(s.db.WithContext(ctx), batch.ID, projectID)
	if err != nil {
		return nil, err
	}
	return s.staffCallBatchView(batch, tickets)
}

type createTicketRequest struct {
	Phone          string `json:"phone"`
	LastName       string `json:"last_name"`
	Honorific      string `json:"honorific"`
	PartySize      *int   `json:"party_size"`
	AllowDuplicate bool   `json:"allow_duplicate"`
}

func (s *Server) createTicket(w http.ResponseWriter, r *http.Request) {
	projectID := r.PathValue("id")
	if err := validateUUID(projectID); err != nil {
		writeError(w, err)
		return
	}
	if err := s.authorizeProject(r.Context(), projectID); err != nil {
		writeError(w, err)
		return
	}
	user := currentPrincipal(r.Context()).User
	s.createTicketForActor(w, r, user.ID, false)
}

func (s *Server) createTicketForActor(w http.ResponseWriter, r *http.Request, actorID string, publicView bool) {
	projectID := r.PathValue("id")
	key, err := readIdempotencyKey(r)
	if err != nil {
		writeError(w, err)
		return
	}
	var input createTicketRequest
	if err := decodeJSON(r, &input); err != nil {
		writeError(w, err)
		return
	}
	if input.PartySize == nil || *input.PartySize < 1 {
		writeError(w, &apiError{Status: http.StatusUnprocessableEntity, Code: "INVALID_PARTY_SIZE", Message: "请选择本号人数"})
		return
	}
	phone, err := security.NormalizePhone(input.Phone)
	if err != nil {
		writeError(w, &apiError{Status: http.StatusUnprocessableEntity, Code: "INVALID_PHONE", Message: err.Error()})
		return
	}
	input.Phone = phone
	input.LastName = strings.TrimSpace(input.LastName)
	if utf8.RuneCountInString(input.LastName) > 40 {
		writeError(w, &apiError{Status: http.StatusUnprocessableEntity, Code: "INVALID_LAST_NAME", Message: "姓氏不能超过 40 个字符"})
		return
	}
	input.Honorific = strings.TrimSpace(input.Honorific)
	if input.Honorific == "" {
		input.Honorific = "游客"
	}
	if input.Honorific != "游客" && input.Honorific != "先生" && input.Honorific != "女士" {
		writeError(w, &apiError{Status: http.StatusUnprocessableEntity, Code: "INVALID_HONORIFIC", Message: "称谓只能是游客、先生或女士"})
		return
	}
	hash, err := requestHash(input)
	if err != nil {
		writeError(w, err)
		return
	}

	scope := "CREATE_TICKET"
	auditAction := "TICKET_CREATED"
	duplicateMessage := "该手机号已有活动号码，请员工确认后继续"
	if publicView {
		scope = "PUBLIC_CREATE_TICKET"
		auditAction = "PUBLIC_TICKET_CREATED"
		duplicateMessage = "该手机号已有活动号码，请确认后继续"
	}
	phoneDigest := s.cipher.Digest(phone)
	if publicView && (s.publicTicketProjectLimiter != nil || s.publicTicketPhoneLimiter != nil) {
		if stored, code, found, err := loadIdempotent(s.db.WithContext(r.Context()), projectID, actorID, scope, key, hash, s.now()); err != nil {
			writeError(w, err)
			return
		} else if found {
			writeRawJSON(w, code, stored)
			return
		}
		if allowed, retry := s.allowPublicTicket(projectID, phoneDigest); !allowed {
			writePublicTicketRateLimit(w, retry)
			return
		}
	}
	var responseBody []byte
	responseCode := http.StatusCreated
	var revision int64
	err = s.db.WithContext(r.Context()).Transaction(func(tx *gorm.DB) error {
		project, session, err := s.lockRunningProjectAndSession(tx, projectID)
		if err != nil {
			return err
		}
		if *input.PartySize < project.MinPartySize || *input.PartySize > project.MaxPartySize {
			return &apiError{
				Status: http.StatusUnprocessableEntity, Code: "INVALID_PARTY_SIZE",
				Message: fmt.Sprintf("本项目每个号码可绑定 %d 到 %d 人", project.MinPartySize, project.MaxPartySize),
				Details: map[string]any{"min_party_size": project.MinPartySize, "max_party_size": project.MaxPartySize},
			}
		}
		if stored, code, found, err := loadIdempotent(tx, projectID, actorID, scope, key, hash, s.now()); err != nil {
			return err
		} else if found {
			responseBody, responseCode = stored, code
			return nil
		}

		var duplicates []model.QueueTicket
		if err := tx.Select("id", "display_number", "status").
			Where("project_id = ? AND queue_session_id = ? AND phone_hmac = ? AND status = ?",
				projectID, session.ID, phoneDigest, model.TicketWaiting).
			Order("joined_at ASC").Find(&duplicates).Error; err != nil {
			return err
		}
		if len(duplicates) > 0 && !input.AllowDuplicate {
			existing := make([]map[string]any, 0, len(duplicates))
			for _, duplicate := range duplicates {
				existing = append(existing, map[string]any{"id": duplicate.ID, "display_number": duplicate.DisplayNumber, "status": duplicate.Status})
			}
			var details any = map[string]any{"tickets": existing}
			if publicView {
				details = nil
			}
			return &apiError{Status: http.StatusConflict, Code: "DUPLICATE_PHONE", Message: duplicateMessage, Details: details}
		}

		publicToken, err := security.GenerateToken()
		if err != nil {
			return err
		}
		displayNumber, err := domain.DisplayNumber("", session.NextTicketNumber)
		if err != nil {
			return err
		}
		phoneCiphertext, phoneNonce, err := s.cipher.Encrypt(phone, []byte("phone:"+projectID))
		if err != nil {
			return err
		}
		var lastNameCiphertext, lastNameNonce []byte
		if input.LastName != "" {
			lastNameCiphertext, lastNameNonce, err = s.cipher.Encrypt(input.LastName, []byte("last_name:"+projectID))
			if err != nil {
				return err
			}
		}
		now := s.now()
		ticket := model.QueueTicket{
			ID: uuid.NewString(), ProjectID: projectID, QueueSessionID: session.ID,
			TicketNumber: session.NextTicketNumber, DisplayNumber: displayNumber, PartySize: *input.PartySize,
			PublicTokenHash: security.HashToken(publicToken), PhoneCiphertext: phoneCiphertext, PhoneNonce: phoneNonce,
			PhoneHMAC: stringPointer(phoneDigest), LastNameCiphertext: lastNameCiphertext, LastNameNonce: lastNameNonce,
			Honorific: input.Honorific, Status: model.TicketWaiting, JoinedAt: now,
			PersonalDataPurgeAt: now.Add(30 * 24 * time.Hour), CreatedBy: actorID, CreatedAt: now, UpdatedAt: now,
		}
		if err := tx.Create(&ticket).Error; err != nil {
			return err
		}
		revision = session.Revision + 1
		if err := tx.Model(&model.QueueSession{}).Where("id = ? AND project_id = ?", session.ID, projectID).
			Updates(map[string]any{"next_ticket_number": session.NextTicketNumber + 1, "revision": revision, "updated_at": now}).Error; err != nil {
			return err
		}
		if err := s.addAudit(tx, r, &projectID, &actorID, auditAction, "QUEUE_TICKET", &ticket.ID,
			map[string]any{"display_number": displayNumber, "party_size": ticket.PartySize, "duplicate_count": len(duplicates), "duplicate_confirmed": input.AllowDuplicate}); err != nil {
			return err
		}
		var ticketResponse map[string]any
		if publicView {
			phoneLast4, err := s.ticketPhoneLast4(ticket)
			if err != nil {
				return err
			}
			ticketResponse = map[string]any{
				"id": ticket.ID, "display_number": ticket.DisplayNumber, "ticket_number": ticket.DisplayNumber,
				"status": ticket.Status, "party_size": ticket.PartySize, "phone_last4": phoneLast4, "created_at": ticket.JoinedAt,
			}
		} else {
			ticketResponse, err = s.staffTicketView(ticket)
			if err != nil {
				return err
			}
		}
		if !publicView {
			ticketResponse["public_token"] = publicToken
			ticketResponse["public_url"] = "/visitor/" + publicToken
		}
		response := map[string]any{
			"ticket": ticketResponse, "public_token": publicToken, "public_url": "/visitor/" + publicToken,
			"status_path": "/api/public/status/" + publicToken, "revision": revision,
		}
		responseBody, err = marshalResponse(response)
		if err != nil {
			return err
		}
		return saveIdempotent(tx, projectID, actorID, scope, key, hash, responseCode, responseBody, &ticket.ID, now)
	})
	if err != nil {
		writeError(w, err)
		return
	}
	if revision > 0 {
		s.hub.publish(queueEvent{ProjectID: projectID, Revision: revision, Type: "ticket.created", At: s.now()})
	}
	writeRawJSON(w, responseCode, responseBody)
}

type callNextRequest struct {
	ExpectedRevision *int64 `json:"expected_revision"`
	Mode             string `json:"mode"`
	Count            *int   `json:"count"`
}

func (s *Server) callNext(w http.ResponseWriter, r *http.Request) {
	projectID := r.PathValue("id")
	if err := validateUUID(projectID); err != nil {
		writeError(w, err)
		return
	}
	if err := s.authorizeProject(r.Context(), projectID); err != nil {
		writeError(w, err)
		return
	}
	var input callNextRequest
	if err := decodeJSON(r, &input); err != nil {
		writeError(w, err)
		return
	}
	if input.ExpectedRevision == nil || *input.ExpectedRevision < 0 {
		writeError(w, &apiError{Status: http.StatusUnprocessableEntity, Code: "EXPECTED_REVISION_REQUIRED", Message: "必须提供当前队列 revision"})
		return
	}
	mode := strings.ToUpper(strings.TrimSpace(input.Mode))
	if mode == "" {
		mode = model.CallModeTicket
	}
	if mode != model.CallModeTicket && mode != model.CallModePeople {
		writeError(w, &apiError{Status: http.StatusUnprocessableEntity, Code: "INVALID_CALL_MODE", Message: "叫号方式不正确"})
		return
	}
	count := 1
	if input.Count != nil {
		count = *input.Count
	}
	if count < 1 || count > 10000 {
		writeError(w, &apiError{Status: http.StatusUnprocessableEntity, Code: "INVALID_CALL_COUNT", Message: "叫号数量必须是有效的正整数"})
		return
	}
	key, err := readIdempotencyKey(r)
	if err != nil {
		writeError(w, err)
		return
	}
	hash, _ := requestHash(map[string]any{"operation": "CALL_NEXT", "project_id": projectID, "expected_revision": *input.ExpectedRevision, "mode": mode, "count": count})
	user := currentPrincipal(r.Context()).User
	var responseBody []byte
	responseCode := http.StatusOK
	var revision int64
	err = s.db.WithContext(r.Context()).Transaction(func(tx *gorm.DB) error {
		project, session, err := s.lockCallableProjectAndSession(tx, projectID)
		if err != nil {
			return err
		}
		if stored, code, found, err := loadIdempotent(tx, projectID, user.ID, "CALL_NEXT", key, hash, s.now()); err != nil {
			return err
		} else if found {
			responseBody, responseCode = stored, code
			return nil
		}
		if session.Revision != *input.ExpectedRevision {
			return &apiError{
				Status: http.StatusConflict, Code: "REVISION_CONFLICT", Message: "队列已更新，请刷新后重试",
				Details: map[string]any{"expected_revision": *input.ExpectedRevision, "current_revision": session.Revision},
			}
		}
		if !projectAllowsCallMode(project.CallMode, mode) {
			return &apiError{Status: http.StatusUnprocessableEntity, Code: "CALL_MODE_NOT_ALLOWED", Message: "当前项目不支持该叫号方式"}
		}
		limit := project.MaxCallTicketCount
		unit := "个号码"
		if mode == model.CallModePeople {
			limit = project.MaxCallPeopleCount
			unit = "人"
		}
		if count > limit {
			return &apiError{
				Status: http.StatusUnprocessableEntity, Code: "CALL_COUNT_EXCEEDS_LIMIT",
				Message: fmt.Sprintf("本次最多可输入 %d %s", limit, unit),
				Details: map[string]any{"mode": mode, "max_count": limit},
			}
		}
		var waitingTickets []model.QueueTicket
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).
			Where("project_id = ? AND queue_session_id = ? AND status = ?", projectID, session.ID, model.TicketWaiting).
			Order("ticket_number ASC").Limit(count).Find(&waitingTickets).Error; err != nil {
			return err
		}
		if len(waitingTickets) == 0 {
			return &apiError{Status: http.StatusConflict, Code: "QUEUE_EMPTY", Message: "当前没有等待中的号码"}
		}
		tickets, peopleCount, err := domain.SelectTicketsForCall(waitingTickets, mode, count)
		if errors.Is(err, domain.ErrFirstTicketExceedsPeopleTarget) {
			return &apiError{
				Status: http.StatusConflict, Code: "NEXT_TICKET_EXCEEDS_PEOPLE_TARGET",
				Message: fmt.Sprintf("队首号码有 %d 人，请输入不少于 %d 人", waitingTickets[0].PartySize, waitingTickets[0].PartySize),
				Details: map[string]any{"next_ticket_number": waitingTickets[0].DisplayNumber, "party_size": waitingTickets[0].PartySize},
			}
		}
		if err != nil {
			return err
		}
		now := s.now()
		autoCompletedBatchIDs, autoCompletedTicketCount, err := autoCompleteActiveBatches(tx, projectID, now)
		if err != nil {
			return err
		}
		var lastSequence int
		if err := tx.Model(&model.CallBatch{}).Where("queue_session_id = ?", session.ID).
			Select("COALESCE(MAX(batch_sequence), 0)").Scan(&lastSequence).Error; err != nil {
			return err
		}
		revision = session.Revision + 1
		batch := model.CallBatch{
			ID: uuid.NewString(), ProjectID: projectID, QueueSessionID: session.ID,
			BatchSequence: lastSequence + 1, Revision: revision, Status: "CALLED",
			CallMode: mode, RequestedCount: count, TicketCount: len(tickets), PeopleCount: peopleCount,
			RequestedBy: user.ID, CalledAt: now, CreatedAt: now,
		}
		if err := tx.Create(&batch).Error; err != nil {
			return err
		}
		links := make([]model.CallBatchTicket, 0, len(tickets))
		ids := make([]string, 0, len(tickets))
		for i := range tickets {
			links = append(links, model.CallBatchTicket{CallBatchID: batch.ID, TicketID: tickets[i].ID, ProjectID: projectID, Position: i + 1, CreatedAt: now})
			ids = append(ids, tickets[i].ID)
			tickets[i].Status = model.TicketCalled
			tickets[i].CalledAt = &now
			tickets[i].UpdatedAt = now
		}
		if err := tx.Create(&links).Error; err != nil {
			return err
		}
		if err := tx.Model(&model.QueueTicket{}).Where("project_id = ? AND id IN ? AND status = ?", projectID, ids, model.TicketWaiting).
			Updates(map[string]any{"status": model.TicketCalled, "called_at": now, "updated_at": now}).Error; err != nil {
			return err
		}
		if err := tx.Model(&model.QueueSession{}).Where("id = ? AND project_id = ?", session.ID, projectID).
			Updates(map[string]any{"revision": revision, "updated_at": now}).Error; err != nil {
			return err
		}
		outcome, detail := simulationResult(project.DeviceSimulationMode)
		simulation := model.DeviceSimulation{
			ID: uuid.NewString(), ProjectID: projectID, CallBatchID: batch.ID, Adapter: "SIMULATOR",
			Outcome: outcome, Detail: detail, AttemptedAt: now, CompletedAt: now, CreatedAt: now,
		}
		if err := tx.Create(&simulation).Error; err != nil {
			return err
		}
		if err := s.addAudit(tx, r, &projectID, &user.ID, "CALL_NEXT", "CALL_BATCH", &batch.ID,
			map[string]any{
				"call_mode": mode, "requested_count": count, "ticket_count": len(tickets), "people_count": peopleCount,
				"revision": revision, "device_outcome": outcome,
				"auto_completed_batch_ids": autoCompletedBatchIDs, "auto_completed_ticket_count": autoCompletedTicketCount,
			}); err != nil {
			return err
		}
		deviceResult := map[string]any{"channel": "SIMULATOR", "status": outcome, "message": detail}
		batchView, err := s.staffCallBatchView(batch, tickets)
		if err != nil {
			return err
		}
		response := map[string]any{
			"batch": batchView, "revision": revision,
			"device":         map[string]any{"adapter": simulation.Adapter, "outcome": outcome, "detail": detail},
			"device_results": []map[string]any{deviceResult},
		}
		responseBody, err = marshalResponse(response)
		if err != nil {
			return err
		}
		return saveIdempotent(tx, projectID, user.ID, "CALL_NEXT", key, hash, responseCode, responseBody, &batch.ID, now)
	})
	if err != nil {
		writeError(w, err)
		return
	}
	if revision > 0 {
		s.hub.publish(queueEvent{ProjectID: projectID, Revision: revision, Type: "batch.called", At: s.now()})
	}
	writeRawJSON(w, responseCode, responseBody)
}

func projectAllowsCallMode(projectMode, requestedMode string) bool {
	return projectMode == model.CallModeBoth || projectMode == requestedMode
}

func autoCompleteActiveBatches(tx *gorm.DB, projectID string, now time.Time) ([]string, int64, error) {
	var batches []model.CallBatch
	if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).
		Where("project_id = ? AND status = 'CALLED'", projectID).
		Order("called_at ASC").Find(&batches).Error; err != nil {
		return nil, 0, err
	}
	if len(batches) == 0 {
		return []string{}, 0, nil
	}
	batchIDs := make([]string, 0, len(batches))
	for _, batch := range batches {
		batchIDs = append(batchIDs, batch.ID)
	}
	var ticketIDs []string
	if err := tx.Model(&model.CallBatchTicket{}).
		Where("project_id = ? AND call_batch_id IN ?", projectID, batchIDs).
		Pluck("ticket_id", &ticketIDs).Error; err != nil {
		return nil, 0, err
	}
	var completedCount int64
	if len(ticketIDs) > 0 {
		result := tx.Model(&model.QueueTicket{}).
			Where("project_id = ? AND id IN ? AND status IN ?", projectID, ticketIDs, []string{model.TicketCalled, model.TicketArrived}).
			Updates(map[string]any{"status": model.TicketCompleted, "completed_at": now, "personal_data_purge_at": now.Add(30 * 24 * time.Hour), "updated_at": now})
		if result.Error != nil {
			return nil, 0, result.Error
		}
		completedCount = result.RowsAffected
	}
	if err := tx.Model(&model.CallBatch{}).
		Where("project_id = ? AND id IN ? AND status = 'CALLED'", projectID, batchIDs).
		Updates(map[string]any{"status": "COMPLETED", "completed_at": now}).Error; err != nil {
		return nil, 0, err
	}
	return batchIDs, completedCount, nil
}

func (s *Server) transitionTicket(w http.ResponseWriter, r *http.Request) {
	ticketID := r.PathValue("id")
	if err := validateUUID(ticketID); err != nil {
		writeError(w, err)
		return
	}
	action := r.PathValue("action")
	target := map[string]string{"arrive": model.TicketArrived, "complete": model.TicketCompleted, "miss": model.TicketMissed}[action]
	if target == "" {
		writeError(w, &apiError{Status: http.StatusNotFound, Code: "ACTION_NOT_FOUND", Message: "操作不存在"})
		return
	}
	var initial model.QueueTicket
	if err := s.db.WithContext(r.Context()).Select("id", "project_id", "queue_session_id", "status").First(&initial, "id = ?", ticketID).Error; err != nil {
		writeError(w, mapNotFound(err, "TICKET_NOT_FOUND", "号码不存在"))
		return
	}
	if err := s.authorizeProject(r.Context(), initial.ProjectID); err != nil {
		writeError(w, err)
		return
	}
	user := currentPrincipal(r.Context()).User
	var ticket model.QueueTicket
	var ticketResponse map[string]any
	var revision int64
	err := s.db.WithContext(r.Context()).Transaction(func(tx *gorm.DB) error {
		var project model.Project
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&project, "id = ?", initial.ProjectID).Error; err != nil {
			return err
		}
		var session model.QueueSession
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&session, "id = ? AND project_id = ?", initial.QueueSessionID, initial.ProjectID).Error; err != nil {
			return err
		}
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&ticket, "id = ? AND project_id = ?", ticketID, initial.ProjectID).Error; err != nil {
			return err
		}
		if !domain.CanTransitionTicket(ticket.Status, target) {
			return &apiError{Status: http.StatusConflict, Code: "INVALID_TICKET_TRANSITION", Message: "当前号码状态不能执行该操作", Details: map[string]any{"status": ticket.Status, "target": target}}
		}
		if ticket.Status == target {
			revision = session.Revision
			var err error
			ticketResponse, err = s.staffTicketView(ticket)
			return err
		}
		now := s.now()
		updates := map[string]any{"status": target, "updated_at": now}
		switch target {
		case model.TicketArrived:
			updates["arrived_at"] = now
			ticket.ArrivedAt = &now
		case model.TicketCompleted:
			updates["completed_at"] = now
			updates["personal_data_purge_at"] = now.Add(30 * 24 * time.Hour)
			ticket.CompletedAt = &now
		case model.TicketMissed:
			updates["missed_at"] = now
			updates["personal_data_purge_at"] = now.Add(30 * 24 * time.Hour)
			ticket.MissedAt = &now
		}
		if err := tx.Model(&model.QueueTicket{}).Where("id = ? AND project_id = ?", ticket.ID, ticket.ProjectID).Updates(updates).Error; err != nil {
			return err
		}
		ticket.Status, ticket.UpdatedAt = target, now
		revision = session.Revision + 1
		if err := tx.Model(&model.QueueSession{}).Where("id = ? AND project_id = ?", session.ID, initial.ProjectID).
			Updates(map[string]any{"revision": revision, "updated_at": now}).Error; err != nil {
			return err
		}
		if target == model.TicketCompleted || target == model.TicketMissed {
			if err := completeBatchIfTerminal(tx, ticket.ID, initial.ProjectID, now); err != nil {
				return err
			}
		}
		if err := s.addAudit(tx, r, &initial.ProjectID, &user.ID, "TICKET_"+target, "QUEUE_TICKET", &ticket.ID,
			map[string]any{"from": initial.Status, "to": target, "revision": revision}); err != nil {
			return err
		}
		var err error
		ticketResponse, err = s.staffTicketView(ticket)
		return err
	})
	if err != nil {
		writeError(w, err)
		return
	}
	s.hub.publish(queueEvent{ProjectID: initial.ProjectID, Revision: revision, Type: "ticket." + strings.ToLower(target), At: s.now()})
	writeJSON(w, http.StatusOK, map[string]any{"ticket": ticketResponse, "revision": revision})
}

func (s *Server) lockRunningProjectAndSession(tx *gorm.DB, projectID string) (model.Project, model.QueueSession, error) {
	return s.lockProjectAndSession(tx, projectID, false)
}

func (s *Server) lockCallableProjectAndSession(tx *gorm.DB, projectID string) (model.Project, model.QueueSession, error) {
	return s.lockProjectAndSession(tx, projectID, true)
}

func (s *Server) lockProjectAndSession(tx *gorm.DB, projectID string, allowPaused bool) (model.Project, model.QueueSession, error) {
	var project model.Project
	if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&project, "id = ?", projectID).Error; err != nil {
		return project, model.QueueSession{}, mapNotFound(err, "PROJECT_NOT_FOUND", "项目不存在")
	}
	if project.Status != model.ProjectRunning && (!allowPaused || project.Status != model.ProjectPaused) {
		return project, model.QueueSession{}, &apiError{Status: http.StatusConflict, Code: "PROJECT_NOT_RUNNING", Message: "项目当前未运行，不能写入队列"}
	}
	businessDate, err := s.businessDateFor(project)
	if err != nil {
		return project, model.QueueSession{}, err
	}
	var session model.QueueSession
	err = tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("project_id = ? AND business_date = ?", projectID, businessDate).First(&session).Error
	if errors.Is(err, gorm.ErrRecordNotFound) {
		now := s.now()
		session = model.QueueSession{
			ID: uuid.NewString(), ProjectID: projectID, BusinessDate: businessDate, Status: "RUNNING",
			NextTicketNumber: 1, Revision: 0, OpenedAt: now, CreatedAt: now, UpdatedAt: now,
		}
		if err := tx.Create(&session).Error; err != nil {
			return project, session, err
		}
		return project, session, nil
	}
	if err != nil {
		return project, session, err
	}
	if session.Status != "RUNNING" && (!allowPaused || session.Status != "PAUSED") {
		return project, session, &apiError{Status: http.StatusConflict, Code: "QUEUE_NOT_RUNNING", Message: "队列当前未运行"}
	}
	return project, session, nil
}

func loadIdempotent(tx *gorm.DB, projectID, userID, scope, key, hash string, now time.Time) ([]byte, int, bool, error) {
	var record model.IdempotencyKey
	err := tx.Where("project_id = ? AND user_id = ? AND scope = ? AND key = ?", projectID, userID, scope, key).First(&record).Error
	if errors.Is(err, gorm.ErrRecordNotFound) {
		return nil, 0, false, nil
	}
	if err != nil {
		return nil, 0, false, err
	}
	if record.RequestHash != hash {
		return nil, 0, false, &apiError{Status: http.StatusConflict, Code: "IDEMPOTENCY_KEY_REUSED", Message: "该幂等键已用于不同请求"}
	}
	return []byte(record.ResponseBody), record.ResponseCode, true, nil
}

func saveIdempotent(tx *gorm.DB, projectID, userID, scope, key, hash string, code int, body []byte, resourceID *string, now time.Time) error {
	record := model.IdempotencyKey{
		ID: uuid.NewString(), ProjectID: projectID, UserID: userID, Scope: scope, Key: key,
		RequestHash: hash, ResponseCode: code, ResponseBody: json.RawMessage(body), ResourceID: resourceID,
		ExpiresAt: now.Add(24 * time.Hour), CreatedAt: now,
	}
	return tx.Create(&record).Error
}

func readIdempotencyKey(r *http.Request) (string, error) {
	key := strings.TrimSpace(r.Header.Get("Idempotency-Key"))
	if len(key) < 8 || len(key) > 128 || strings.ContainsAny(key, " \t\r\n") {
		return "", &apiError{Status: http.StatusBadRequest, Code: "IDEMPOTENCY_KEY_REQUIRED", Message: "Idempotency-Key 必须为 8 到 128 个不含空白的字符"}
	}
	return key, nil
}

func writeRawJSON(w http.ResponseWriter, status int, body []byte) {
	w.Header().Set("Content-Type", "application/json; charset=utf-8")
	w.WriteHeader(status)
	_, _ = w.Write(body)
}

func validateUUID(value string) error {
	if _, err := uuid.Parse(value); err != nil {
		return &apiError{Status: http.StatusBadRequest, Code: "INVALID_ID", Message: "资源 ID 格式不正确"}
	}
	return nil
}

func mapNotFound(err error, code, message string) error {
	if errors.Is(err, gorm.ErrRecordNotFound) {
		return &apiError{Status: http.StatusNotFound, Code: code, Message: message}
	}
	return err
}

func (s *Server) staffTicketView(ticket model.QueueTicket) (map[string]any, error) {
	phone, lastName, err := s.decryptTicketPersonal(ticket)
	if err != nil {
		return nil, err
	}
	return map[string]any{
		"id": ticket.ID, "display_number": ticket.DisplayNumber, "ticket_number": ticket.DisplayNumber, "status": ticket.Status,
		"party_size":              ticket.PartySize,
		"reissued_from_ticket_id": ticket.ReissuedFromTicketID,
		"phone":                   phone, "last_name": lastName, "honorific": ticket.Honorific,
		"joined_at": ticket.JoinedAt, "called_at": ticket.CalledAt,
		"created_at": ticket.JoinedAt,
		"arrived_at": ticket.ArrivedAt, "completed_at": ticket.CompletedAt, "missed_at": ticket.MissedAt,
	}, nil
}

func (s *Server) staffCallBatchView(batch model.CallBatch, tickets []model.QueueTicket) (map[string]any, error) {
	views := make([]map[string]any, 0, len(tickets))
	for _, ticket := range tickets {
		view, err := s.staffTicketView(ticket)
		if err != nil {
			return nil, err
		}
		views = append(views, view)
	}
	return map[string]any{
		"id": batch.ID, "sequence": batch.BatchSequence, "batch_number": batch.BatchSequence, "status": batch.Status,
		"call_mode": batch.CallMode, "requested_count": batch.RequestedCount,
		"ticket_count": batch.TicketCount, "people_count": batch.PeopleCount,
		"called_at": batch.CalledAt, "revision": batch.Revision, "tickets": views,
	}, nil
}

func batchTickets(db *gorm.DB, batchID, projectID string) ([]model.QueueTicket, error) {
	var tickets []model.QueueTicket
	err := db.Model(&model.QueueTicket{}).
		Joins("JOIN call_batch_tickets cbt ON cbt.ticket_id = queue_tickets.id AND cbt.project_id = queue_tickets.project_id").
		Where("cbt.call_batch_id = ? AND cbt.project_id = ?", batchID, projectID).
		Order("cbt.position ASC").Find(&tickets).Error
	return tickets, err
}

func simulationResult(mode string) (string, string) {
	switch mode {
	case "FAILURE":
		return "FAILURE", "模拟设备执行失败；业务叫号已成功"
	case "DISABLED":
		return "SKIPPED", "设备模拟器已停用；业务叫号已成功"
	default:
		return "SUCCESS", "模拟设备执行成功"
	}
}

func completeBatchIfTerminal(tx *gorm.DB, ticketID, projectID string, now time.Time) error {
	var link model.CallBatchTicket
	err := tx.Where("ticket_id = ? AND project_id = ?", ticketID, projectID).First(&link).Error
	if errors.Is(err, gorm.ErrRecordNotFound) {
		return nil
	}
	if err != nil {
		return err
	}
	var active int64
	if err := tx.Model(&model.QueueTicket{}).
		Joins("JOIN call_batch_tickets cbt ON cbt.ticket_id = queue_tickets.id AND cbt.project_id = queue_tickets.project_id").
		Where("cbt.call_batch_id = ? AND cbt.project_id = ? AND queue_tickets.status IN ?", link.CallBatchID, projectID,
			[]string{model.TicketCalled, model.TicketArrived}).Count(&active).Error; err != nil {
		return err
	}
	if active == 0 {
		return tx.Model(&model.CallBatch{}).Where("id = ? AND project_id = ?", link.CallBatchID, projectID).
			Updates(map[string]any{"status": "COMPLETED", "completed_at": now}).Error
	}
	return nil
}