package store import ( "context" "database/sql" "encoding/json" "strconv" "strings" "time" ) type AICapability struct { Feature string `db:"feature"` Enabled bool `db:"enabled"` Model string `db:"model"` DailyLimit int `db:"daily_limit"` ConcurrencyLimit int `db:"concurrency_limit"` MaxInputChars int `db:"max_input_chars"` MaxFileSizeMB int `db:"max_file_size_mb"` ConfigJSON string `db:"config_json"` } type AITask struct { TaskID string `db:"task_id"` TenantCode string `db:"tenant_code"` UserID string `db:"user_id"` Feature string `db:"feature"` Status string `db:"status"` Progress int `db:"progress"` SourceType string `db:"source_type"` SourceRefID string `db:"source_ref_id"` InputHash string `db:"input_hash"` PromptVersion string `db:"prompt_version"` Model string `db:"model"` ResultRefID string `db:"result_ref_id"` ResultJSON sql.NullString `db:"result_json"` ErrorReason string `db:"error_reason"` ErrorMessage string `db:"error_message"` RetryCount int `db:"retry_count"` StartedAt *time.Time `db:"started_at"` FinishedAt *time.Time `db:"finished_at"` CreatedAt time.Time `db:"created_at"` UpdatedAt time.Time `db:"updated_at"` } type AIRecord struct { RecordID string `db:"record_id"` TenantCode string `db:"tenant_code"` UserID string `db:"user_id"` RecordType string `db:"record_type"` Title string `db:"title"` Summary string `db:"summary"` Status string `db:"status"` DetailID string `db:"detail_id"` TaskID string `db:"task_id"` CreatedAt time.Time `db:"created_at"` UpdatedAt time.Time `db:"updated_at"` DeletedAt *time.Time `db:"deleted_at"` } type AIDocument struct { RecordID string `db:"record_id"` UserID string `db:"user_id"` DocumentType string `db:"document_type"` Title string `db:"title"` ContentMarkdown string `db:"content_markdown"` ContentPlain string `db:"content_plain"` SectionsJSON string `db:"sections_json"` SourceExcerpt string `db:"source_excerpt"` PromptVersion string `db:"prompt_version"` Model string `db:"model"` CreatedAt time.Time `db:"created_at"` UpdatedAt time.Time `db:"updated_at"` } type AITableAnalysis struct { RecordID string `db:"record_id"` UserID string `db:"user_id"` FileID string `db:"file_id"` FileName string `db:"file_name"` FileType string `db:"file_type"` FileSize int64 `db:"file_size"` Topic string `db:"topic"` MetricsJSON string `db:"metrics_json"` PreviewJSON string `db:"preview_json"` AlertsJSON string `db:"alerts_json"` Interpretation string `db:"interpretation"` ExportFileID string `db:"export_file_id"` PromptVersion string `db:"prompt_version"` Model string `db:"model"` CreatedAt time.Time `db:"created_at"` UpdatedAt time.Time `db:"updated_at"` } type AIMeetingSummary struct { SummaryID string `db:"summary_id"` MeetingID string `db:"meeting_id"` UserID string `db:"user_id"` TaskID string `db:"task_id"` ConclusionsJSON string `db:"conclusions_json"` DecisionsJSON string `db:"decisions_json"` ActionItemsJSON string `db:"action_items_json"` SourceFileIDsJSON string `db:"source_file_ids_json"` PromptVersion string `db:"prompt_version"` Model string `db:"model"` EditedBy string `db:"edited_by"` EditedAt *time.Time `db:"edited_at"` Version int `db:"version"` CreatedAt time.Time `db:"created_at"` UpdatedAt time.Time `db:"updated_at"` } type AIUsageLog struct { TenantCode string UserID string Feature string TaskID string Provider string Model string ProviderRequestID string PromptTokens int CompletionTokens int TotalTokens int LatencyMS int Status string ErrorReason string InputHash string OutputHash string } type AIFileMetadata struct { MIMEType string FileSize int64 FileETag string Location string } type CompletedFileUpload struct { ID int64 UserID int64 MIMEType string Size int64 ETag string Location string CreatedAt time.Time } func (s *Store) AICapabilities(ctx context.Context) ([]AICapability, error) { var items []AICapability err := s.db.SelectContext(ctx, &items, `SELECT feature,enabled,model,daily_limit,concurrency_limit, max_input_chars,max_file_size_mb,COALESCE(CAST(config_json AS CHAR), '{}') AS config_json FROM wb_ai_capability_config WHERE tenant_id=? ORDER BY feature ASC`, s.tenant()) if err != nil { return nil, err } if len(items) > 0 { return items, nil } err = s.db.SelectContext(ctx, &items, `SELECT feature,enabled,model,daily_limit,concurrency_limit, max_input_chars,max_file_size_mb,COALESCE(CAST(config_json AS CHAR), '{}') AS config_json FROM wb_ai_capability_config WHERE tenant_id=0 ORDER BY feature ASC`) return items, err } func (s *Store) AIFileMetadata(ctx context.Context, fileID string) AIFileMetadata { meta := s.fileUploadMetadata(ctx, strings.TrimSpace(fileID)) return AIFileMetadata{ MIMEType: meta.MIMEType, FileSize: meta.FileSize, FileETag: meta.FileETag, Location: meta.Location, } } func (s *Store) CreateAITask(ctx context.Context, task AITask, result any) error { now := time.Now().UTC().Round(time.Millisecond) resultJSON, err := marshalNullable(result) if err != nil { return err } _, 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,?,?,?,?)`, 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, resultJSON, task.ErrorReason, task.ErrorMessage, task.StartedAt, task.FinishedAt, now, now) return err } 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) if err != nil { return 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=?`, status, progress, resultRefID, resultJSON, reason, message, now, now, s.tenant(), taskID) if err != nil { return err } if rows, _ := res.RowsAffected(); rows == 0 { return ErrNotFound } return 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=? WHERE tenant_id=? AND user_id=? AND task_id=? AND status IN ('queued','processing')`, now, now, s.tenant(), formatInt64(userID), taskID) if err != nil { return err } if rows, _ := res.RowsAffected(); rows == 0 { return ErrNotFound } return nil } 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, 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 { return nil, ErrNotFound } if err != nil { return nil, err } return &task, nil } func (s *Store) InsertAIUsageLog(ctx context.Context, log AIUsageLog) error { now := time.Now().UTC().Round(time.Millisecond) _, err := s.db.ExecContext(ctx, `INSERT INTO wb_ai_usage_log (tenant_id,tenant_code,user_id,feature,task_id,provider,model,provider_request_id,prompt_tokens, completion_tokens,total_tokens,latency_ms,status,error_reason,input_hash,output_hash,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, s.tenant(), log.TenantCode, log.UserID, log.Feature, log.TaskID, log.Provider, log.Model, log.ProviderRequestID, log.PromptTokens, log.CompletionTokens, log.TotalTokens, log.LatencyMS, log.Status, log.ErrorReason, log.InputHash, log.OutputHash, now) return err } func (s *Store) SaveAIDocument(ctx context.Context, tenantCode string, userID int64, record AIRecord, doc AIDocument) error { now := time.Now().UTC().Round(time.Millisecond) tx, err := s.db.BeginTxx(ctx, nil) if err != nil { return err } defer tx.Rollback() _, err = tx.ExecContext(ctx, `INSERT INTO wb_ai_record (record_id,tenant_id,tenant_code,user_id,record_type,title,summary,status,detail_id,task_id,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)`, record.RecordID, s.tenant(), tenantCode, formatInt64(userID), record.RecordType, record.Title, record.Summary, record.Status, doc.RecordID, record.TaskID, now, now) if err != nil { return err } _, err = tx.ExecContext(ctx, `INSERT INTO wb_ai_document (record_id,tenant_id,user_id,document_type,title,content_markdown,content_plain,sections_json,source_excerpt, prompt_version,model,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)`, doc.RecordID, s.tenant(), formatInt64(userID), doc.DocumentType, doc.Title, doc.ContentMarkdown, doc.ContentPlain, doc.SectionsJSON, doc.SourceExcerpt, doc.PromptVersion, doc.Model, now, now) if err != nil { return err } return tx.Commit() } func (s *Store) AIDocuments(ctx context.Context, userID int64, queryText, cursor string, limit int) ([]AIRecord, string, error) { return s.aiRecords(ctx, userID, "smart_document", queryText, cursor, limit) } func (s *Store) AIDocument(ctx context.Context, userID int64, recordID string) (*AIDocument, error) { var doc AIDocument err := s.db.GetContext(ctx, &doc, `SELECT d.record_id,d.user_id,d.document_type,d.title,d.content_markdown, COALESCE(d.content_plain,'') AS content_plain,COALESCE(CAST(d.sections_json AS CHAR), '[]') AS sections_json, d.source_excerpt,d.prompt_version,d.model,d.created_at,d.updated_at FROM wb_ai_document d JOIN wb_ai_record r ON r.tenant_id=d.tenant_id AND r.record_id=d.record_id WHERE d.tenant_id=? AND d.user_id=? AND d.record_id=? AND r.deleted_at IS NULL`, s.tenant(), formatInt64(userID), recordID) if err == sql.ErrNoRows { return nil, ErrNotFound } if err != nil { return nil, err } return &doc, nil } func (s *Store) SaveAITableAnalysis(ctx context.Context, tenantCode string, userID int64, record AIRecord, analysis AITableAnalysis) error { now := time.Now().UTC().Round(time.Millisecond) tx, err := s.db.BeginTxx(ctx, nil) if err != nil { return err } defer tx.Rollback() _, err = tx.ExecContext(ctx, `INSERT INTO wb_ai_record (record_id,tenant_id,tenant_code,user_id,record_type,title,summary,status,detail_id,task_id,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)`, record.RecordID, s.tenant(), tenantCode, formatInt64(userID), record.RecordType, record.Title, record.Summary, record.Status, analysis.RecordID, record.TaskID, now, now) if err != nil { return err } _, err = tx.ExecContext(ctx, `INSERT INTO wb_ai_table_analysis (record_id,tenant_id,user_id,file_id,file_name,file_type,file_size,topic,metrics_json,preview_json,alerts_json, interpretation,export_file_id,prompt_version,model,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, analysis.RecordID, s.tenant(), formatInt64(userID), analysis.FileID, analysis.FileName, analysis.FileType, analysis.FileSize, analysis.Topic, analysis.MetricsJSON, analysis.PreviewJSON, analysis.AlertsJSON, analysis.Interpretation, analysis.ExportFileID, analysis.PromptVersion, analysis.Model, now, now) if err != nil { return err } return tx.Commit() } func (s *Store) SaveCompletedFileUpload(ctx context.Context, file CompletedFileUpload) error { now := file.CreatedAt.UTC().Round(time.Millisecond) if now.IsZero() { now = time.Now().UTC().Round(time.Millisecond) } _, err := s.db.ExecContext(ctx, `INSERT INTO im_fileuploads (tenant_id,id,createdat,updatedat,userid,status,mimetype,size,etag,location) VALUES(?,?,?,?,?,?,?,?,?,?)`, s.tenant(), file.ID, now, now, formatInt64(file.UserID), 1, file.MIMEType, file.Size, file.ETag, file.Location) return err } func (s *Store) UpdateAITableAnalysisExportFileID(ctx context.Context, userID int64, recordID, fileID string) error { now := time.Now().UTC().Round(time.Millisecond) res, err := s.db.ExecContext(ctx, `UPDATE wb_ai_table_analysis SET export_file_id=?,updated_at=? WHERE tenant_id=? AND user_id=? AND record_id=?`, fileID, now, s.tenant(), formatInt64(userID), recordID) if err != nil { return err } affected, err := res.RowsAffected() if err != nil { return err } if affected == 0 { return ErrNotFound } return nil } func (s *Store) AITableAnalyses(ctx context.Context, userID int64, queryText, cursor string, limit int) ([]AIRecord, string, error) { return s.aiRecords(ctx, userID, "smart_table", queryText, cursor, limit) } func (s *Store) AIRecords(ctx context.Context, userID int64, recordType, queryText, cursor string, limit int) ([]AIRecord, string, error) { return s.aiRecords(ctx, userID, recordType, queryText, cursor, limit) } func (s *Store) DeleteAIRecord(ctx context.Context, userID int64, recordID string) error { now := time.Now().UTC().Round(time.Millisecond) res, err := s.db.ExecContext(ctx, `UPDATE wb_ai_record SET status='deleted',deleted_at=?,updated_at=? WHERE tenant_id=? AND user_id=? AND record_id=? AND deleted_at IS NULL`, now, now, s.tenant(), formatInt64(userID), recordID) if err != nil { return err } if rows, _ := res.RowsAffected(); rows == 0 { return ErrNotFound } return nil } func (s *Store) AITableAnalysis(ctx context.Context, userID int64, recordID string) (*AITableAnalysis, error) { var item AITableAnalysis err := s.db.GetContext(ctx, &item, `SELECT a.record_id,a.user_id,a.file_id,a.file_name,a.file_type,a.file_size,a.topic, COALESCE(CAST(a.metrics_json AS CHAR), '[]') AS metrics_json, COALESCE(CAST(a.preview_json AS CHAR), '{}') AS preview_json, COALESCE(CAST(a.alerts_json AS CHAR), '[]') AS alerts_json, a.interpretation,a.export_file_id,a.prompt_version,a.model,a.created_at,a.updated_at FROM wb_ai_table_analysis a JOIN wb_ai_record r ON r.tenant_id=a.tenant_id AND r.record_id=a.record_id WHERE a.tenant_id=? AND a.user_id=? AND a.record_id=? AND r.deleted_at IS NULL`, s.tenant(), formatInt64(userID), recordID) if err == sql.ErrNoRows { return nil, ErrNotFound } if err != nil { return nil, err } return &item, nil } func (s *Store) UpsertAIMeetingSummary(ctx context.Context, tenantCode string, userID int64, summary AIMeetingSummary) error { now := time.Now().UTC().Round(time.Millisecond) _, err := s.db.ExecContext(ctx, `INSERT INTO wb_ai_meeting_summary (summary_id,tenant_id,tenant_code,meeting_id,user_id,task_id,conclusions_json,decisions_json,action_items_json, source_file_ids_json,prompt_version,model,version,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,1,?,?) ON DUPLICATE KEY UPDATE task_id=VALUES(task_id),conclusions_json=VALUES(conclusions_json), decisions_json=VALUES(decisions_json),action_items_json=VALUES(action_items_json), source_file_ids_json=VALUES(source_file_ids_json),prompt_version=VALUES(prompt_version),model=VALUES(model), version=version+1,updated_at=VALUES(updated_at)`, summary.SummaryID, s.tenant(), tenantCode, summary.MeetingID, formatInt64(userID), summary.TaskID, summary.ConclusionsJSON, summary.DecisionsJSON, summary.ActionItemsJSON, summary.SourceFileIDsJSON, summary.PromptVersion, summary.Model, now, now) return err } func (s *Store) AIMeetingSummary(ctx context.Context, userID int64, meetingID string) (*AIMeetingSummary, error) { var item AIMeetingSummary err := s.db.GetContext(ctx, &item, `SELECT summary_id,meeting_id,user_id,task_id, COALESCE(CAST(conclusions_json AS CHAR), '[]') AS conclusions_json, COALESCE(CAST(decisions_json AS CHAR), '[]') AS decisions_json, COALESCE(CAST(action_items_json AS CHAR), '[]') AS action_items_json, COALESCE(CAST(source_file_ids_json AS CHAR), '[]') AS source_file_ids_json, prompt_version,model,edited_by,edited_at,version,created_at,updated_at FROM wb_ai_meeting_summary WHERE tenant_id=? AND meeting_id=? LIMIT 1`, s.tenant(), meetingID) if err == sql.ErrNoRows { return nil, ErrNotFound } if err != nil { return nil, err } return &item, nil } func (s *Store) UpdateAIMeetingSummary(ctx context.Context, userID int64, meetingID, summaryID, conclusionsJSON, decisionsJSON, actionItemsJSON string) (*AIMeetingSummary, error) { meeting, err := s.Meeting(ctx, userID, meetingID) if err != nil { return nil, err } if meeting.OrganizerID != userID && meeting.RecorderID != userID { return nil, ErrForbidden } now := time.Now().UTC().Round(time.Millisecond) res, err := s.db.ExecContext(ctx, `UPDATE wb_ai_meeting_summary SET conclusions_json=?,decisions_json=?,action_items_json=?, edited_by=?,edited_at=?,version=version+1,updated_at=? WHERE tenant_id=? AND meeting_id=? AND summary_id=?`, conclusionsJSON, decisionsJSON, actionItemsJSON, formatInt64(userID), now, now, s.tenant(), meetingID, summaryID) if err != nil { return nil, err } if rows, _ := res.RowsAffected(); rows == 0 { return nil, ErrNotFound } return s.AIMeetingSummary(ctx, userID, meetingID) } func (s *Store) aiRecords(ctx context.Context, userID int64, recordType, queryText, cursor string, limit int) ([]AIRecord, string, error) { where := `tenant_id=? AND user_id=? AND deleted_at IS NULL` args := []any{s.tenant(), formatInt64(userID)} if recordType != "" { where += ` AND record_type=?` args = append(args, recordType) } if queryText != "" { where += ` AND (title LIKE ? OR summary LIKE ?)` like := "%" + queryText + "%" args = append(args, like, like) } if cursor != "" { where += ` AND record_id limit { next = items[limit-1].RecordID items = items[:limit] } return items, next, nil } func marshalNullable(value any) (any, error) { if value == nil { return nil, nil } data, err := json.Marshal(value) if err != nil { return nil, err } return string(data), nil } func formatInt64(value int64) string { return strconv.FormatInt(value, 10) }