From d2e70efd9d01559fa3339c5f55dabce7b231e8b6 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Wed, 30 Sep 2026 11:37:49 +0800 Subject: [PATCH] feat: async ai workbench tasks --- internal/app/app.go | 10 +- internal/httpapi/ai.go | 143 ++++--------- internal/httpapi/ai_worker.go | 366 ++++++++++++++++++++++++++++++++++ internal/store/ai.go | 78 +++++++- 4 files changed, 493 insertions(+), 104 deletions(-) create mode 100644 internal/httpapi/ai_worker.go diff --git a/internal/app/app.go b/internal/app/app.go index cf71aee..7e2b834 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -21,6 +21,7 @@ type App struct { redis *redis.Client events events.Publisher api *httpapi.API + cancel context.CancelFunc } func New(cfg *config.Config) (*App, error) { @@ -60,11 +61,15 @@ func New(cfg *config.Config) (*App, error) { return nil, err } aiService := ai.NewService(ai.NewClient(cfg.AI)) + api := httpapi.New(st, auth.New(cfg.Auth, redisClient), publisher, cfg.Auth, cfg.Manager, aiService) + ctx, cancel := context.WithCancel(context.Background()) + api.StartAIWorkers(ctx) return &App{ store: st, redis: redisClient, events: publisher, - api: httpapi.New(st, auth.New(cfg.Auth, redisClient), publisher, cfg.Auth, cfg.Manager, aiService), + api: api, + cancel: cancel, }, nil } @@ -77,6 +82,9 @@ func (a *App) Handler() http.Handler { } func (a *App) Close() { + if a.cancel != nil { + a.cancel() + } if a.redis != nil { _ = a.redis.Close() } diff --git a/internal/httpapi/ai.go b/internal/httpapi/ai.go index 709d3ef..97aa385 100644 --- a/internal/httpapi/ai.go +++ b/internal/httpapi/ai.go @@ -118,48 +118,28 @@ func (a *API) aiDocumentGenerate(w http.ResponseWriter, r *http.Request, p *auth return } taskID := "aitask_" + randomToken(12) - started := time.Now().UTC().Round(time.Millisecond) + snapshot, err := aiTaskSnapshot(aiDocumentTaskSnapshot{ + SourceText: source, + TitleHint: strings.TrimSpace(body.TitleHint), + DocumentType: strings.TrimSpace(body.DocumentType), + Language: strings.TrimSpace(body.Language), + }) + if err != nil { + writeStoreError(w, r, err) + return + } if err := st.CreateAITask(r.Context(), store.AITask{ TaskID: taskID, TenantCode: p.TenantCode, UserID: strconv.FormatInt(p.UserID, 10), Feature: "smart_document", - Status: "processing", Progress: 10, SourceType: "text", InputHash: hashText(source), + Status: "queued", Progress: 0, SourceType: "text", InputHash: hashText(source), + InputSnapshot: snapshot, PromptVersion: "smart_document_v1", Model: "", - StartedAt: &started, }, nil); err != nil { writeStoreError(w, r, err) return } - result, err := a.ai.GenerateDocument(r.Context(), source, strings.TrimSpace(body.TitleHint), strings.TrimSpace(body.DocumentType), strings.TrimSpace(body.Language)) - if err != nil { - _ = st.UpdateAITaskResult(r.Context(), taskID, "failed", 100, "", nil, aiErrorReason(err), err.Error()) - _ = st.InsertAIUsageLog(r.Context(), store.AIUsageLog{ - TenantCode: p.TenantCode, UserID: formatUserID(p.UserID), Feature: "smart_document", TaskID: taskID, - Provider: "deepseek", Status: "failed", ErrorReason: aiErrorReason(err), InputHash: hashText(source), - }) - a.writeAIError(w, r, err) - return - } - taskResult := map[string]any{ - "title": result.Title, - "summary": result.Summary, - "content_markdown": result.ContentMarkdown, - "content_plain": result.ContentPlain, - "sections": result.Sections, - "source_excerpt": excerpt(source, 200), - "can_save": true, - } - if err := st.UpdateAITaskResult(r.Context(), taskID, "succeeded", 100, "", taskResult, "", ""); err != nil { - writeStoreError(w, r, err) - return - } - _ = st.InsertAIUsageLog(r.Context(), store.AIUsageLog{ - TenantCode: p.TenantCode, UserID: formatUserID(p.UserID), Feature: "smart_document", TaskID: taskID, - Provider: "deepseek", Model: result.Usage.Model, ProviderRequestID: result.Usage.ProviderRequestID, - PromptTokens: result.Usage.PromptTokens, CompletionTokens: result.Usage.CompletionTokens, TotalTokens: result.Usage.TotalTokens, - LatencyMS: int(result.Usage.LatencyMS), Status: "succeeded", InputHash: hashText(source), OutputHash: hashText(result.ContentMarkdown), - }) writeSuccess(w, r, map[string]any{ "task_id": taskID, - "status": "succeeded", + "status": "queued", "polling_interval_ms": 1500, "expires_at": time.Now().UTC().Add(2 * time.Hour).Format(time.RFC3339Nano), }) @@ -179,6 +159,13 @@ func (a *API) aiTaskResource(w http.ResponseWriter, r *http.Request, p *auth.Pri writeStoreError(w, r, err) return } + feature := "" + if task, err := st.AITask(r.Context(), p.UserID, taskID); err == nil { + feature = task.Feature + } + a.publishAITaskEvent(r.Context(), p.TenantID, p.TenantCode, p.UserID, taskID, "ai", "AI 任务已取消", "", map[string]any{ + "task_id": taskID, "feature": feature, "status": "canceled", "progress": 100, "result_type": "none", + }) writeSuccess(w, r, map[string]any{"task_id": taskID, "status": "canceled", "canceled_at": time.Now().UTC().Format(time.RFC3339Nano)}) return } @@ -370,50 +357,23 @@ func (a *API) aiTableAnalysisCreate(w http.ResponseWriter, r *http.Request, p *a } taskID := "aitask_tbl_" + randomToken(12) recordID := "airec_tbl_" + randomToken(10) - started := time.Now().UTC().Round(time.Millisecond) + snapshot, err := aiTaskSnapshot(aiTableTaskSnapshot{ + FileID: fileID, FileName: strings.TrimSpace(body.FileName), FileType: fileType, FileSize: fileSize, + Topic: topic, SheetName: strings.TrimSpace(body.SheetName), Language: strings.TrimSpace(body.Language), RecordID: recordID, + }) + if err != nil { + writeStoreError(w, r, err) + return + } if err := st.CreateAITask(r.Context(), store.AITask{ TaskID: taskID, TenantCode: p.TenantCode, UserID: strconv.FormatInt(p.UserID, 10), Feature: "smart_table", - Status: "processing", Progress: 20, SourceType: "file", SourceRefID: fileID, InputHash: hashText(fileID + ":" + topic), - PromptVersion: "smart_table_v1", StartedAt: &started, + Status: "queued", Progress: 0, SourceType: "file", SourceRefID: fileID, InputHash: hashText(fileID + ":" + topic), + InputSnapshot: snapshot, PromptVersion: "smart_table_v1", }, nil); err != nil { writeStoreError(w, r, err) return } - profile := a.tableProfileForAI(r, fileID, fileMeta, fileType) - result, err := a.ai.AnalyzeTable(r.Context(), ai.TableAnalysisInput{ - Topic: topic, FileID: fileID, FileName: strings.TrimSpace(body.FileName), FileType: fileType, FileSize: fileSize, Profile: profile, - }) - if err != nil { - _ = st.UpdateAITaskResult(r.Context(), taskID, "failed", 100, "", nil, aiErrorReason(err), err.Error()) - a.writeAIError(w, r, err) - return - } - metricsJSON := mustJSON(result.Metrics, "[]") - previewJSON := mustJSON(result.Preview, "{}") - alertsJSON := mustJSON(result.Alerts, "[]") - if err := st.SaveAITableAnalysis(r.Context(), p.TenantCode, p.UserID, store.AIRecord{ - RecordID: recordID, RecordType: "smart_table", Title: result.Title, Summary: excerpt(firstNonEmpty(result.Summary, result.Interpretation), 160), - Status: "succeeded", TaskID: taskID, - }, store.AITableAnalysis{ - RecordID: recordID, FileID: fileID, FileName: strings.TrimSpace(body.FileName), FileType: fileType, FileSize: fileSize, - Topic: topic, MetricsJSON: metricsJSON, PreviewJSON: previewJSON, AlertsJSON: alertsJSON, Interpretation: result.Interpretation, - PromptVersion: "smart_table_v1", Model: result.Usage.Model, - }); err != nil { - writeStoreError(w, r, err) - return - } - taskResult := map[string]any{"record_id": recordID, "title": result.Title, "metrics": result.Metrics, "alerts": result.Alerts, "interpretation": result.Interpretation} - if err := st.UpdateAITaskResult(r.Context(), taskID, "succeeded", 100, recordID, taskResult, "", ""); err != nil { - writeStoreError(w, r, err) - return - } - _ = st.InsertAIUsageLog(r.Context(), store.AIUsageLog{ - TenantCode: p.TenantCode, UserID: formatUserID(p.UserID), Feature: "smart_table", TaskID: taskID, - Provider: "deepseek", Model: result.Usage.Model, ProviderRequestID: result.Usage.ProviderRequestID, - PromptTokens: result.Usage.PromptTokens, CompletionTokens: result.Usage.CompletionTokens, TotalTokens: result.Usage.TotalTokens, - LatencyMS: int(result.Usage.LatencyMS), Status: "succeeded", InputHash: hashText(fileID + ":" + topic), OutputHash: hashText(result.Interpretation), - }) - writeSuccess(w, r, map[string]any{"task_id": taskID, "record_id": recordID, "status": "succeeded", "polling_interval_ms": 1500}) + writeSuccess(w, r, map[string]any{"task_id": taskID, "record_id": recordID, "status": "queued", "polling_interval_ms": 1500}) } func (a *API) aiTableAnalysisResource(w http.ResponseWriter, r *http.Request, p *auth.Principal, st *store.Store) { @@ -532,8 +492,7 @@ func (a *API) aiMeetingSummaryCreate(w http.ResponseWriter, r *http.Request, p * if _, ok := a.requireAICapability(w, r, st, "meeting_summary"); !ok { return } - meeting, err := st.Meeting(r.Context(), p.UserID, meetingID) - if err != nil { + if _, err := st.Meeting(r.Context(), p.UserID, meetingID); err != nil { writeStoreError(w, r, err) return } @@ -542,40 +501,26 @@ func (a *API) aiMeetingSummaryCreate(w http.ResponseWriter, r *http.Request, p * writeStoreError(w, r, err) return } - fileNames, fileIDs := meetingSummaryFiles(files, body.FileIDs) + _, fileIDs := meetingSummaryFiles(files, body.FileIDs) taskID := "aitask_mtg_" + randomToken(12) summaryID := "aisum_" + randomToken(10) - started := time.Now().UTC().Round(time.Millisecond) + snapshot, err := aiTaskSnapshot(aiMeetingSummaryTaskSnapshot{ + MeetingID: meetingID, SummaryID: summaryID, FileIDs: fileIDs, + SummaryType: strings.TrimSpace(body.SummaryType), ForceRegenerate: body.ForceRegenerate, + }) + if err != nil { + writeStoreError(w, r, err) + return + } if err := st.CreateAITask(r.Context(), store.AITask{ TaskID: taskID, TenantCode: p.TenantCode, UserID: strconv.FormatInt(p.UserID, 10), Feature: "meeting_summary", - Status: "processing", Progress: 20, SourceType: "meeting", SourceRefID: meetingID, InputHash: hashText(meetingID + strings.Join(fileIDs, ",")), - PromptVersion: "meeting_summary_v1", StartedAt: &started, + Status: "queued", Progress: 0, SourceType: "meeting", SourceRefID: meetingID, InputHash: hashText(meetingID + strings.Join(fileIDs, ",")), + InputSnapshot: snapshot, PromptVersion: "meeting_summary_v1", }, nil); err != nil { writeStoreError(w, r, err) return } - result, err := a.ai.SummarizeMeeting(r.Context(), ai.MeetingSummaryInput{ - MeetingTitle: meeting.Title, StartAt: meeting.StartAt.UTC().Format(time.RFC3339Nano), EndAt: meeting.EndAt.UTC().Format(time.RFC3339Nano), - Participants: meetingSummaryParticipants(*meeting), Files: fileNames, - }) - if err != nil { - _ = st.UpdateAITaskResult(r.Context(), taskID, "failed", 100, "", nil, aiErrorReason(err), err.Error()) - a.writeAIError(w, r, err) - return - } - if err := st.UpsertAIMeetingSummary(r.Context(), p.TenantCode, p.UserID, store.AIMeetingSummary{ - SummaryID: summaryID, MeetingID: meetingID, TaskID: taskID, - ConclusionsJSON: mustJSON(result.Conclusions, "[]"), DecisionsJSON: mustJSON(result.Decisions, "[]"), - ActionItemsJSON: mustJSON(result.ActionItems, "[]"), SourceFileIDsJSON: mustJSON(fileIDs, "[]"), - PromptVersion: "meeting_summary_v1", Model: result.Usage.Model, - }); err != nil { - writeStoreError(w, r, err) - return - } - _ = st.UpdateAITaskResult(r.Context(), taskID, "succeeded", 100, summaryID, map[string]any{ - "summary_id": summaryID, "meeting_id": meetingID, "conclusions": result.Conclusions, "decisions": result.Decisions, "action_items": result.ActionItems, - }, "", "") - writeSuccess(w, r, map[string]any{"task_id": taskID, "meeting_id": meetingID, "status": "succeeded", "polling_interval_ms": 1500}) + writeSuccess(w, r, map[string]any{"task_id": taskID, "meeting_id": meetingID, "summary_id": summaryID, "status": "queued", "polling_interval_ms": 1500}) } func (a *API) aiMeetingSummaryUpdate(w http.ResponseWriter, r *http.Request, p *auth.Principal, st *store.Store, meetingID string) { diff --git a/internal/httpapi/ai_worker.go b/internal/httpapi/ai_worker.go new file mode 100644 index 0000000..b48ef20 --- /dev/null +++ b/internal/httpapi/ai_worker.go @@ -0,0 +1,366 @@ +package httpapi + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "log" + "strconv" + "strings" + "time" + + "git.118im.com/im_team/im_workbench/internal/ai" + "git.118im.com/im_team/im_workbench/internal/events" + "git.118im.com/im_team/im_workbench/internal/store" +) + +const ( + aiWorkerCount = 2 + aiWorkerPollInterval = time.Second + aiTaskRunTimeout = 3 * time.Minute +) + +type aiDocumentTaskSnapshot struct { + SourceText string `json:"source_text"` + TitleHint string `json:"title_hint"` + DocumentType string `json:"document_type"` + Language string `json:"language"` +} + +type aiTableTaskSnapshot struct { + FileID string `json:"file_id"` + FileName string `json:"file_name"` + FileType string `json:"file_type"` + FileSize int64 `json:"file_size"` + Topic string `json:"topic"` + SheetName string `json:"sheet_name"` + Language string `json:"language"` + RecordID string `json:"record_id"` +} + +type aiMeetingSummaryTaskSnapshot struct { + MeetingID string `json:"meeting_id"` + SummaryID string `json:"summary_id"` + FileIDs []string `json:"file_ids"` + SummaryType string `json:"summary_type"` + ForceRegenerate bool `json:"force_regenerate"` +} + +func (a *API) StartAIWorkers(ctx context.Context) { + if a == nil || a.store == nil || a.ai == nil { + return + } + for i := 0; i < aiWorkerCount; i++ { + go a.aiWorkerLoop(ctx, i+1) + } +} + +func (a *API) aiWorkerLoop(ctx context.Context, workerID int) { + ticker := time.NewTicker(aiWorkerPollInterval) + defer ticker.Stop() + for { + task, err := a.store.ClaimQueuedAITask(ctx) + if errors.Is(err, store.ErrNotFound) { + select { + case <-ctx.Done(): + return + case <-ticker.C: + continue + } + } + if err != nil { + log.Printf("ai worker claim failed: worker=%d err=%v", workerID, err) + select { + case <-ctx.Done(): + return + case <-ticker.C: + continue + } + } + a.runAITask(ctx, task) + } +} + +func (a *API) runAITask(parent context.Context, task *store.AITask) { + ctx, cancel := context.WithTimeout(parent, aiTaskRunTimeout) + defer cancel() + st, err := a.store.ForTenant(ctx, task.TenantID, task.TenantCode) + if err != nil { + log.Printf("ai task tenant store failed: task_id=%s tenant_id=%d err=%v", task.TaskID, task.TenantID, err) + return + } + switch task.Feature { + case "smart_document": + a.runAIDocumentTask(ctx, st, task) + case "smart_table": + a.runAITableTask(ctx, st, task) + case "meeting_summary": + a.runAIMeetingSummaryTask(ctx, st, task) + default: + a.finishAITaskFailure(ctx, st, task, "AI_TASK_UNSUPPORTED", "unsupported ai task feature") + } +} + +func (a *API) runAIDocumentTask(ctx context.Context, st *store.Store, task *store.AITask) { + var input aiDocumentTaskSnapshot + if !decodeTaskSnapshot(task, &input) { + a.finishAITaskFailure(ctx, st, task, reasonMalformedRequest, "invalid task input") + return + } + result, err := a.ai.GenerateDocument(ctx, strings.TrimSpace(input.SourceText), strings.TrimSpace(input.TitleHint), strings.TrimSpace(input.DocumentType), strings.TrimSpace(input.Language)) + if err != nil { + a.finishAITaskAIError(ctx, st, task, err, hashText(input.SourceText)) + return + } + taskResult := map[string]any{ + "title": result.Title, + "summary": result.Summary, + "content_markdown": result.ContentMarkdown, + "content_plain": result.ContentPlain, + "sections": result.Sections, + "source_excerpt": excerpt(input.SourceText, 200), + "can_save": true, + } + ok, err := st.FinishAITaskResult(ctx, task.TaskID, "succeeded", 100, "", taskResult, "", "") + if err != nil { + log.Printf("ai document task finish failed: task_id=%s err=%v", task.TaskID, err) + return + } + if !ok { + return + } + userID := taskUserID(task) + _ = st.InsertAIUsageLog(ctx, store.AIUsageLog{ + TenantCode: task.TenantCode, UserID: task.UserID, Feature: "smart_document", TaskID: task.TaskID, + Provider: "deepseek", Model: result.Usage.Model, ProviderRequestID: result.Usage.ProviderRequestID, + PromptTokens: result.Usage.PromptTokens, CompletionTokens: result.Usage.CompletionTokens, TotalTokens: result.Usage.TotalTokens, + LatencyMS: int(result.Usage.LatencyMS), Status: "succeeded", InputHash: hashText(input.SourceText), OutputHash: hashText(result.ContentMarkdown), + }) + a.publishAITaskEvent(ctx, task.TenantID, task.TenantCode, userID, task.TaskID, "smart_document", result.Title, result.Summary, map[string]any{ + "task_id": task.TaskID, "feature": "smart_document", "status": "succeeded", "progress": 100, "result_type": "text", + "result": compactAITextResult(result.Title, result.Summary, result.ContentMarkdown, result.ContentPlain), + "updated_at": time.Now().UTC().Format(time.RFC3339Nano), + }) +} + +func (a *API) runAITableTask(ctx context.Context, st *store.Store, task *store.AITask) { + var input aiTableTaskSnapshot + if !decodeTaskSnapshot(task, &input) { + a.finishAITaskFailure(ctx, st, task, reasonMalformedRequest, "invalid task input") + return + } + fileType := strings.ToLower(strings.TrimSpace(input.FileType)) + fileMeta := st.AIFileMetadata(ctx, input.FileID) + profile := a.tableProfileForAIContext(ctx, input.FileID, fileMeta, fileType) + result, err := a.ai.AnalyzeTable(ctx, ai.TableAnalysisInput{ + Topic: input.Topic, FileID: input.FileID, FileName: input.FileName, FileType: fileType, FileSize: input.FileSize, Profile: profile, + }) + if err != nil { + a.finishAITaskAIError(ctx, st, task, err, hashText(input.FileID+":"+input.Topic)) + return + } + recordID := strings.TrimSpace(input.RecordID) + if recordID == "" { + recordID = "airec_tbl_" + randomToken(10) + } + if err := st.SaveAITableAnalysis(ctx, task.TenantCode, taskUserID(task), store.AIRecord{ + RecordID: recordID, RecordType: "smart_table", Title: result.Title, Summary: excerpt(firstNonEmpty(result.Summary, result.Interpretation), 160), + Status: "succeeded", TaskID: task.TaskID, + }, store.AITableAnalysis{ + RecordID: recordID, FileID: input.FileID, FileName: input.FileName, FileType: fileType, FileSize: input.FileSize, + Topic: input.Topic, MetricsJSON: mustJSON(result.Metrics, "[]"), PreviewJSON: mustJSON(result.Preview, "{}"), + AlertsJSON: mustJSON(result.Alerts, "[]"), Interpretation: result.Interpretation, + PromptVersion: "smart_table_v1", Model: result.Usage.Model, + }); err != nil { + a.finishAITaskFailure(ctx, st, task, reasonAITaskFailed, err.Error()) + return + } + taskResult := map[string]any{"record_id": recordID, "title": result.Title, "metrics": result.Metrics, "alerts": result.Alerts, "interpretation": result.Interpretation} + ok, err := st.FinishAITaskResult(ctx, task.TaskID, "succeeded", 100, recordID, taskResult, "", "") + if err != nil { + log.Printf("ai table task finish failed: task_id=%s err=%v", task.TaskID, err) + return + } + if !ok { + return + } + _ = st.InsertAIUsageLog(ctx, store.AIUsageLog{ + TenantCode: task.TenantCode, UserID: task.UserID, Feature: "smart_table", TaskID: task.TaskID, + Provider: "deepseek", Model: result.Usage.Model, ProviderRequestID: result.Usage.ProviderRequestID, + PromptTokens: result.Usage.PromptTokens, CompletionTokens: result.Usage.CompletionTokens, TotalTokens: result.Usage.TotalTokens, + LatencyMS: int(result.Usage.LatencyMS), Status: "succeeded", InputHash: hashText(input.FileID + ":" + input.Topic), OutputHash: hashText(result.Interpretation), + }) + a.publishAITaskEvent(ctx, task.TenantID, task.TenantCode, taskUserID(task), task.TaskID, "smart_table", result.Title, result.Summary, map[string]any{ + "task_id": task.TaskID, "feature": "smart_table", "status": "succeeded", "progress": 100, "result_type": "record", + "record_id": recordID, + "result": map[string]any{"record_id": recordID, "title": result.Title, "summary": excerpt(firstNonEmpty(result.Summary, result.Interpretation), 160)}, + "updated_at": time.Now().UTC().Format(time.RFC3339Nano), + }) +} + +func (a *API) runAIMeetingSummaryTask(ctx context.Context, st *store.Store, task *store.AITask) { + var input aiMeetingSummaryTaskSnapshot + if !decodeTaskSnapshot(task, &input) { + a.finishAITaskFailure(ctx, st, task, reasonMalformedRequest, "invalid task input") + return + } + userID := taskUserID(task) + meeting, err := st.Meeting(ctx, userID, input.MeetingID) + if err != nil { + a.finishAITaskFailure(ctx, st, task, reasonAITaskFailed, err.Error()) + return + } + files, err := st.MeetingNoteFiles(ctx, userID, input.MeetingID) + if err != nil { + a.finishAITaskFailure(ctx, st, task, reasonAITaskFailed, err.Error()) + return + } + fileNames, fileIDs := meetingSummaryFiles(files, input.FileIDs) + result, err := a.ai.SummarizeMeeting(ctx, ai.MeetingSummaryInput{ + MeetingTitle: meeting.Title, StartAt: meeting.StartAt.UTC().Format(time.RFC3339Nano), EndAt: meeting.EndAt.UTC().Format(time.RFC3339Nano), + Participants: meetingSummaryParticipants(*meeting), Files: fileNames, + }) + if err != nil { + a.finishAITaskAIError(ctx, st, task, err, hashText(input.MeetingID+strings.Join(fileIDs, ","))) + return + } + summaryID := strings.TrimSpace(input.SummaryID) + if summaryID == "" { + summaryID = "aisum_" + randomToken(10) + } + if err := st.UpsertAIMeetingSummary(ctx, task.TenantCode, userID, store.AIMeetingSummary{ + SummaryID: summaryID, MeetingID: input.MeetingID, TaskID: task.TaskID, + ConclusionsJSON: mustJSON(result.Conclusions, "[]"), DecisionsJSON: mustJSON(result.Decisions, "[]"), + ActionItemsJSON: mustJSON(result.ActionItems, "[]"), SourceFileIDsJSON: mustJSON(fileIDs, "[]"), + PromptVersion: "meeting_summary_v1", Model: result.Usage.Model, + }); err != nil { + a.finishAITaskFailure(ctx, st, task, reasonAITaskFailed, err.Error()) + return + } + taskResult := map[string]any{ + "summary_id": summaryID, "meeting_id": input.MeetingID, + "conclusions": result.Conclusions, "decisions": result.Decisions, "action_items": result.ActionItems, + } + ok, err := st.FinishAITaskResult(ctx, task.TaskID, "succeeded", 100, summaryID, taskResult, "", "") + if err != nil { + log.Printf("ai meeting summary task finish failed: task_id=%s err=%v", task.TaskID, err) + return + } + if !ok { + return + } + _ = st.InsertAIUsageLog(ctx, store.AIUsageLog{ + TenantCode: task.TenantCode, UserID: task.UserID, Feature: "meeting_summary", TaskID: task.TaskID, + Provider: "deepseek", Model: result.Usage.Model, ProviderRequestID: result.Usage.ProviderRequestID, + PromptTokens: result.Usage.PromptTokens, CompletionTokens: result.Usage.CompletionTokens, TotalTokens: result.Usage.TotalTokens, + LatencyMS: int(result.Usage.LatencyMS), Status: "succeeded", InputHash: hashText(input.MeetingID + strings.Join(fileIDs, ",")), OutputHash: hashText(mustJSON(result.Conclusions, "[]")), + }) + a.publishAITaskEvent(ctx, task.TenantID, task.TenantCode, userID, task.TaskID, "meeting_summary", meeting.Title, "", map[string]any{ + "task_id": task.TaskID, "feature": "meeting_summary", "status": "succeeded", "progress": 100, "result_type": "summary", + "meeting_id": input.MeetingID, "summary_id": summaryID, + "result": map[string]any{"meeting_id": input.MeetingID, "summary_id": summaryID, "conclusions": result.Conclusions, "decisions": result.Decisions, "action_items": result.ActionItems}, + "updated_at": time.Now().UTC().Format(time.RFC3339Nano), + }) +} + +func (a *API) finishAITaskAIError(ctx context.Context, st *store.Store, task *store.AITask, err error, inputHash string) { + reason := aiErrorReason(err) + a.finishAITaskFailure(ctx, st, task, reason, err.Error()) + _ = st.InsertAIUsageLog(ctx, store.AIUsageLog{ + TenantCode: task.TenantCode, UserID: task.UserID, Feature: task.Feature, TaskID: task.TaskID, + Provider: "deepseek", Status: "failed", ErrorReason: reason, InputHash: inputHash, + }) +} + +func (a *API) finishAITaskFailure(ctx context.Context, st *store.Store, task *store.AITask, reason, message string) { + ok, err := st.FinishAITaskResult(ctx, task.TaskID, "failed", 100, "", nil, reason, message) + if err != nil { + log.Printf("ai task failure update failed: task_id=%s reason=%s err=%v", task.TaskID, reason, err) + return + } + if !ok { + return + } + a.publishAITaskEvent(ctx, task.TenantID, task.TenantCode, taskUserID(task), task.TaskID, task.Feature, "AI 任务失败", message, map[string]any{ + "task_id": task.TaskID, "feature": task.Feature, "status": "failed", "progress": 100, "result_type": "none", + "error": map[string]any{"reason": reason, "message": message}, + "updated_at": time.Now().UTC().Format(time.RFC3339Nano), + }) +} + +func (a *API) publishAITaskEvent(ctx context.Context, tenantID int64, tenantCode string, userID int64, taskID, feature, title, summary string, data map[string]any) { + if a == nil || a.events == nil || userID <= 0 || strings.TrimSpace(taskID) == "" { + return + } + event := events.Event{ + EventID: events.StableEventID(tenantID, "ai", "workbench.ai.task.updated", taskID+":"+stringValue(data["status"]), userID), + TenantID: tenantID, + TenantCode: tenantCode, + ActorUserID: userID, + Module: "ai", + Action: "workbench.ai.task.updated", + ResourceID: taskID, + TargetUserIDs: []int64{userID}, + Title: strings.TrimSpace(title), + Summary: strings.TrimSpace(summary), + OccurredAt: time.Now().UTC().Round(time.Millisecond), + Data: data, + } + pubCtx, cancel := context.WithTimeout(ctx, workbenchEventPublishTimeout) + defer cancel() + if err := a.events.Publish(pubCtx, event); err != nil { + log.Printf("ai task event publish failed: event_id=%s task_id=%s feature=%s err=%v", event.EventID, taskID, feature, err) + } +} + +func aiTaskSnapshot(value any) (sql.NullString, error) { + data, err := json.Marshal(value) + if err != nil { + return sql.NullString{}, err + } + return sql.NullString{String: string(data), Valid: true}, nil +} + +func decodeTaskSnapshot(task *store.AITask, dst any) bool { + if task == nil || !task.InputSnapshot.Valid || strings.TrimSpace(task.InputSnapshot.String) == "" { + return false + } + return json.Unmarshal([]byte(task.InputSnapshot.String), dst) == nil +} + +func (a *API) tableProfileForAIContext(ctx context.Context, fileID string, meta store.AIFileMetadata, fileType string) ai.TableProfile { + if path := resolveTableFilePath(fileID, meta.Location); path != "" { + return tableProfileFromPath(path, fileType) + } + if strings.TrimSpace(meta.Location) == "" { + return ai.TableProfile{Parsed: false, Message: "source file metadata is not available"} + } + path, cleanup, err := a.downloadOSSObjectToTemp(ctx, meta.Location, fileType) + if err != nil { + return ai.TableProfile{Parsed: false, Message: "source file is not available: " + err.Error()} + } + defer cleanup() + return tableProfileFromPath(path, fileType) +} + +func taskUserID(task *store.AITask) int64 { + if task == nil { + return 0 + } + userID, _ := strconv.ParseInt(strings.TrimSpace(task.UserID), 10, 64) + return userID +} + +func compactAITextResult(title, summary, markdown, plain string) map[string]any { + result := map[string]any{"title": title, "summary": summary, "content_plain": plain} + if len([]rune(markdown)) <= 8000 { + result["content_markdown"] = markdown + } + return result +} + +func stringValue(value any) string { + text, _ := value.(string) + return strings.TrimSpace(text) +} diff --git a/internal/store/ai.go b/internal/store/ai.go index 75c060e..e4d415b 100644 --- a/internal/store/ai.go +++ b/internal/store/ai.go @@ -22,6 +22,7 @@ type AICapability struct { type AITask struct { TaskID string `db:"task_id"` + TenantID int64 `db:"tenant_id"` TenantCode string `db:"tenant_code"` UserID string `db:"user_id"` Feature string `db:"feature"` @@ -30,6 +31,7 @@ type AITask struct { SourceType string `db:"source_type"` SourceRefID string `db:"source_ref_id"` InputHash string `db:"input_hash"` + InputSnapshot sql.NullString `db:"input_snapshot"` PromptVersion string `db:"prompt_version"` Model string `db:"model"` ResultRefID string `db:"result_ref_id"` @@ -181,17 +183,59 @@ func (s *Store) CreateAITask(ctx context.Context, task AITask, result any) error if err != nil { return err } + var inputSnapshot any + if task.InputSnapshot.Valid { + inputSnapshot = task.InputSnapshot.String + } _, err = s.db.ExecContext(ctx, `INSERT INTO wb_ai_task (task_id,tenant_id,tenant_code,user_id,feature,status,progress,source_type,source_ref_id,input_hash, input_snapshot,prompt_version,model,result_ref_id,result_json,error_reason,error_message,retry_count, started_at,finished_at,created_at,updated_at) - VALUES(?,?,?,?,?,?,?,?,?,?,NULL,?,?,?,?,?,?,0,?,?,?,?)`, + VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,0,?,?,?,?)`, task.TaskID, s.tenant(), task.TenantCode, task.UserID, task.Feature, task.Status, task.Progress, - task.SourceType, task.SourceRefID, task.InputHash, task.PromptVersion, task.Model, task.ResultRefID, + task.SourceType, task.SourceRefID, task.InputHash, inputSnapshot, task.PromptVersion, task.Model, task.ResultRefID, resultJSON, task.ErrorReason, task.ErrorMessage, task.StartedAt, task.FinishedAt, now, now) return err } +func (s *Store) ClaimQueuedAITask(ctx context.Context) (*AITask, error) { + tx, err := s.db.BeginTxx(ctx, nil) + if err != nil { + return nil, err + } + defer tx.Rollback() + var task AITask + err = tx.GetContext(ctx, &task, `SELECT task_id,tenant_id,tenant_code,user_id,feature,status,progress,source_type,source_ref_id, + input_hash,input_snapshot,prompt_version,model,result_ref_id,result_json,error_reason,error_message,retry_count, + started_at,finished_at,created_at,updated_at + FROM wb_ai_task + WHERE status='queued' AND feature IN ('smart_document','smart_table','meeting_summary') + ORDER BY created_at ASC,id ASC LIMIT 1 FOR UPDATE SKIP LOCKED`) + if err == sql.ErrNoRows { + return nil, ErrNotFound + } + if err != nil { + return nil, err + } + now := time.Now().UTC().Round(time.Millisecond) + res, err := tx.ExecContext(ctx, `UPDATE wb_ai_task SET status='processing',progress=?,started_at=COALESCE(started_at,?),updated_at=? + WHERE tenant_id=? AND task_id=? AND status='queued'`, maxInt(task.Progress, 10), now, now, task.TenantID, task.TaskID) + if err != nil { + return nil, err + } + if rows, _ := res.RowsAffected(); rows == 0 { + return nil, ErrNotFound + } + if err := tx.Commit(); err != nil { + return nil, err + } + task.Status = "processing" + task.Progress = maxInt(task.Progress, 10) + task.StartedAt = &now + task.UpdatedAt = now + return &task, nil +} + func (s *Store) UpdateAITaskResult(ctx context.Context, taskID, status string, progress int, resultRefID string, result any, reason, message string) error { now := time.Now().UTC().Round(time.Millisecond) resultJSON, err := marshalNullable(result) @@ -210,6 +254,25 @@ func (s *Store) UpdateAITaskResult(ctx context.Context, taskID, status string, p return nil } +func (s *Store) FinishAITaskResult(ctx context.Context, taskID, status string, progress int, resultRefID string, result any, reason, message string) (bool, error) { + now := time.Now().UTC().Round(time.Millisecond) + resultJSON, err := marshalNullable(result) + if err != nil { + return false, err + } + res, err := s.db.ExecContext(ctx, `UPDATE wb_ai_task SET status=?,progress=?,result_ref_id=?,result_json=?, + error_reason=?,error_message=?,finished_at=?,updated_at=? WHERE tenant_id=? AND task_id=? AND status='processing'`, + status, progress, resultRefID, resultJSON, reason, message, now, now, s.tenant(), taskID) + if err != nil { + return false, err + } + rows, err := res.RowsAffected() + if err != nil { + return false, err + } + return rows > 0, nil +} + func (s *Store) CancelAITask(ctx context.Context, userID int64, taskID string) error { now := time.Now().UTC().Round(time.Millisecond) res, err := s.db.ExecContext(ctx, `UPDATE wb_ai_task SET status='canceled',progress=100,finished_at=?,updated_at=? @@ -226,8 +289,8 @@ func (s *Store) CancelAITask(ctx context.Context, userID int64, taskID string) e func (s *Store) AITask(ctx context.Context, userID int64, taskID string) (*AITask, error) { var task AITask - err := s.db.GetContext(ctx, &task, `SELECT task_id,tenant_code,user_id,feature,status,progress,source_type,source_ref_id, - input_hash,prompt_version,model,result_ref_id,result_json,error_reason,error_message,retry_count, + err := s.db.GetContext(ctx, &task, `SELECT task_id,tenant_id,tenant_code,user_id,feature,status,progress,source_type,source_ref_id, + input_hash,input_snapshot,prompt_version,model,result_ref_id,result_json,error_reason,error_message,retry_count, started_at,finished_at,created_at,updated_at FROM wb_ai_task WHERE tenant_id=? AND user_id=? AND task_id=?`, s.tenant(), formatInt64(userID), taskID) if err == sql.ErrNoRows { @@ -527,3 +590,10 @@ func marshalNullable(value any) (any, error) { func formatInt64(value int64) string { return strconv.FormatInt(value, 10) } + +func maxInt(a, b int) int { + if a > b { + return a + } + return b +}