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 } var session model.QueueSession err := s.db.WithContext(r.Context()).Where("project_id = ? AND status IN ?", projectID, []string{"RUNNING", "PAUSED"}). Order("business_date DESC").First(&session).Error 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 = "该手机号已有活动号码,请确认后继续" } 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 } phoneDigest := s.cipher.Digest(phone) var duplicates []model.QueueTicket if err := tx.Select("id", "display_number", "status"). Where("project_id = ? AND phone_hmac = ? AND status IN ?", projectID, phoneDigest, []string{model.TicketWaiting, model.TicketCalled, model.TicketArrived}). 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.lockRunningProjectAndSession(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) { 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 { return project, model.QueueSession{}, &apiError{Status: http.StatusConflict, Code: "PROJECT_NOT_RUNNING", Message: "项目当前未运行,不能写入队列"} } location, err := time.LoadLocation(project.Timezone) if err != nil { return project, model.QueueSession{}, fmt.Errorf("invalid project timezone: %w", err) } businessDate := s.now().In(location).Format("2006-01-02") 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() date, _ := time.ParseInLocation("2006-01-02", businessDate, location) session = model.QueueSession{ ID: uuid.NewString(), ProjectID: projectID, BusinessDate: date, 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" { 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 }