Files
olons_server/internal/store/ai.go
T

530 lines
20 KiB
Go

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"`
FileName string `db:"file_name"`
FileType string `db:"file_type"`
FileSize int64 `db:"file_size"`
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) {
where := `r.tenant_id=? AND r.user_id=? AND r.record_type='smart_table' AND r.deleted_at IS NULL`
args := []any{s.tenant(), formatInt64(userID)}
if queryText != "" {
where += ` AND (r.title LIKE ? OR r.summary LIKE ? OR a.file_name LIKE ?)`
like := "%" + queryText + "%"
args = append(args, like, like, like)
}
if cursor != "" {
where += ` AND r.record_id<?`
args = append(args, cursor)
}
args = append(args, limit+1)
var items []AIRecord
err := s.db.SelectContext(ctx, &items, `SELECT r.record_id,r.tenant_code,r.user_id,r.record_type,r.title,r.summary,r.status,r.detail_id,r.task_id,
COALESCE(a.file_name, '') AS file_name,COALESCE(a.file_type, '') AS file_type,COALESCE(a.file_size, 0) AS file_size,
r.created_at,r.updated_at,r.deleted_at
FROM wb_ai_record r
LEFT JOIN wb_ai_table_analysis a ON a.tenant_id=r.tenant_id AND a.record_id=r.detail_id
WHERE `+where+` ORDER BY r.created_at DESC,r.record_id DESC LIMIT ?`, args...)
if err != nil {
return nil, "", err
}
next := ""
if len(items) > limit {
next = items[limit-1].RecordID
items = items[:limit]
}
return items, next, nil
}
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<?`
args = append(args, cursor)
}
args = append(args, limit+1)
var items []AIRecord
err := s.db.SelectContext(ctx, &items, `SELECT record_id,tenant_code,user_id,record_type,title,summary,status,detail_id,task_id,
created_at,updated_at,deleted_at FROM wb_ai_record WHERE `+where+` ORDER BY created_at DESC,record_id DESC LIMIT ?`, args...)
if err != nil {
return nil, "", err
}
next := ""
if len(items) > 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)
}