package app import ( "context" "database/sql" "errors" "fmt" "log" "net/http" "os" "path/filepath" "strconv" "strings" "time" ) const ( chatMediaCleanupLock = "im_chat_media_cleanup_v1" chatMediaBatchSize = 200 ) type chatMediaCleanupSummary struct { Scanned int `json:"scanned"` Deleted int `json:"deleted"` Failed int `json:"failed"` } type chatMediaTarget struct { MessageID int64 MediaAssetID sql.NullInt64 MediaType string PublicURL string ConversationID int64 Seq int64 StorageProvider string Bucket string ObjectKey string AssetStatus int } type chatMediaEvent struct { MessageID int64 ConversationID int64 Seq int64 } func clampChatMediaRetentionDays(days int) int { if days < 1 || days > 3650 { return 90 } return days } func chatMediaObjectKeys(objectKey, mediaType string) []string { objectKey = strings.TrimSpace(objectKey) if objectKey == "" { return nil } keys := []string{objectKey} if mediaType == "image" { extension := filepath.Ext(objectKey) stem := strings.TrimSuffix(objectKey, extension) // Chat/post images uploaded by this service use the immutable -im1 // suffix and always have a derived thumbnail sibling. if strings.HasSuffix(stem, "-im1") { keys = append(keys, stem+"-thumb.jpg") } } return keys } func chatMediaCleanupError(err error) string { if err == nil { return "" } message := strings.TrimSpace(err.Error()) runes := []rune(message) if len(runes) > 480 { message = string(runes[:480]) } return message } func isMissingMediaObject(err error) bool { if err == nil || errors.Is(err, os.ErrNotExist) { return true } message := strings.ToLower(err.Error()) return strings.Contains(message, "nosuchkey") || strings.Contains(message, "no such key") || strings.Contains(message, "not found") || strings.Contains(message, "status code: 404") || strings.Contains(message, "statuscode=404") } func (a *App) adminChatMedia(w http.ResponseWriter, r *http.Request) { page, size, offset := pagination(r) keyword := strings.TrimSpace(r.URL.Query().Get("keyword")) mediaType := strings.TrimSpace(r.URL.Query().Get("type")) statusText := strings.TrimSpace(r.URL.Query().Get("status")) where := ` WHERE 1=1` args := []any{} if mediaType == "image" || mediaType == "voice" { where += ` AND mm.media_type=?` args = append(args, mediaType) } if statusText != "" { status, err := strconv.Atoi(statusText) if err == nil && status >= 0 && status <= 3 { where += ` AND mm.status=?` args = append(args, status) } } if keyword != "" { like := "%" + keyword + "%" where += ` AND (sp.nickname LIKE ? OR su.public_id LIKE ? OR mm.public_url LIKE ? OR CAST(mm.message_id AS CHAR) LIKE ? OR CAST(m.conversation_id AS CHAR) LIKE ?)` args = append(args, like, like, like, like, like) } base := ` FROM im_message_media mm JOIN im_messages m ON m.id=mm.message_id JOIN users su ON su.id=m.sender_id JOIN user_profiles sp ON sp.user_id=m.sender_id LEFT JOIN media_assets ma ON ma.id=mm.media_asset_id` var total int64 if err := a.db.QueryRowContext(r.Context(), `SELECT COUNT(*)`+base+where, args...).Scan(&total); err != nil { fail(w, http.StatusInternalServerError, 50001, "查询聊天媒体失败") return } queryArgs := append(append([]any{}, args...), size, offset) rows, err := a.db.QueryContext(r.Context(), `SELECT mm.message_id,mm.media_asset_id,m.conversation_id,m.seq,m.sender_id,su.public_id,sp.nickname,sp.avatar_url,mm.media_type,mm.public_url,mm.duration_ms,mm.status,mm.delete_reason,mm.cleanup_error,mm.deleted_at,mm.created_at,COALESCE(ma.storage_provider,''),COALESCE(ma.bucket,''),COALESCE(ma.object_key,''),COALESCE(ma.mime_type,''),COALESCE(ma.file_size,0)`+base+where+` ORDER BY mm.created_at DESC,mm.message_id DESC LIMIT ? OFFSET ?`, queryArgs...) if err != nil { fail(w, http.StatusInternalServerError, 50001, "查询聊天媒体失败") return } defer rows.Close() items := []map[string]any{} for rows.Next() { var messageID, conversationID, seq, senderID, fileSize int64 var assetID sql.NullInt64 var durationMS sql.NullInt64 var deletedAt sql.NullTime var publicID, nickname, avatar, typ, publicURL, reason, cleanupError, provider, bucket, objectKey, mimeType string var status int var createdAt time.Time if err = rows.Scan(&messageID, &assetID, &conversationID, &seq, &senderID, &publicID, &nickname, &avatar, &typ, &publicURL, &durationMS, &status, &reason, &cleanupError, &deletedAt, &createdAt, &provider, &bucket, &objectKey, &mimeType, &fileSize); err != nil { fail(w, http.StatusInternalServerError, 50001, "读取聊天媒体失败") return } items = append(items, map[string]any{ "id": messageID, "messageId": messageID, "mediaAssetId": nullableInt64(assetID), "conversationId": conversationID, "seq": seq, "senderId": senderID, "senderPublicId": publicID, "senderNickname": nickname, "senderAvatar": avatar, "type": typ, "url": publicURL, "durationMs": nullableInt64(durationMS), "status": status, "deleteReason": reason, "cleanupError": cleanupError, "deletedAt": nullableTime(deletedAt), "createdAt": createdAt, "storageProvider": provider, "bucket": bucket, "objectKey": objectKey, "mimeType": mimeType, "fileSize": fileSize, }) } if err = rows.Err(); err != nil { fail(w, http.StatusInternalServerError, 50001, "读取聊天媒体失败") return } stats := map[string]any{"total": 0, "active": 0, "deleted": 0, "failed": 0, "activeBytes": 0} var statsTotal, active, deleted, failed, activeBytes int64 if scanErr := a.db.QueryRowContext(r.Context(), `SELECT COUNT(*),COALESCE(SUM(mm.status=1),0),COALESCE(SUM(mm.status=0),0),COALESCE(SUM(mm.status=3),0),COALESCE(SUM(IF(mm.status=1,ma.file_size,0)),0) FROM im_message_media mm LEFT JOIN media_assets ma ON ma.id=mm.media_asset_id`).Scan(&statsTotal, &active, &deleted, &failed, &activeBytes); scanErr == nil { stats = map[string]any{"total": statsTotal, "active": active, "deleted": deleted, "failed": failed, "activeBytes": activeBytes} } reply(w, map[string]any{ "items": items, "total": total, "page": page, "size": size, "stats": stats, "retention": map[string]any{ "enabled": a.configBool(r.Context(), "im.chat_media_retention_enabled", false), "days": clampChatMediaRetentionDays(a.configInt(r.Context(), "im.chat_media_retention_days", 90)), "lastAt": a.configPlain(r.Context(), "im.chat_media_cleanup_last_at", ""), "lastResult": a.configPlain(r.Context(), "im.chat_media_cleanup_last_result", ""), }, }) } func (a *App) adminUpdateChatMediaRetention(w http.ResponseWriter, r *http.Request) { var req struct { Enabled bool `json:"enabled"` Days int `json:"days"` } if decode(r, &req) != nil || req.Days < 1 || req.Days > 3650 { fail(w, http.StatusBadRequest, 20001, "保留天数应为 1 至 3650 天") return } tx, err := a.db.BeginTx(r.Context(), nil) if err == nil { _, err = tx.ExecContext(r.Context(), `UPDATE system_configs SET config_value=? WHERE config_key='im.chat_media_retention_enabled'`, strconv.FormatBool(req.Enabled)) } if err == nil { _, err = tx.ExecContext(r.Context(), `UPDATE system_configs SET config_value=? WHERE config_key='im.chat_media_retention_days'`, strconv.Itoa(req.Days)) } if err != nil || tx.Commit() != nil { if tx != nil { _ = tx.Rollback() } fail(w, http.StatusInternalServerError, 50001, "保存聊天媒体清理策略失败") return } a.audit(r, "update_retention", "chat_media", 0, map[string]any{"enabled": req.Enabled, "days": req.Days}) reply(w, map[string]any{"enabled": req.Enabled, "days": req.Days}) } func (a *App) adminCleanupDueChatMedia(w http.ResponseWriter, r *http.Request) { days := clampChatMediaRetentionDays(a.configInt(r.Context(), "im.chat_media_retention_days", 90)) result, acquired, err := a.withChatMediaCleanupLock(r.Context(), func(ctx context.Context) (chatMediaCleanupSummary, error) { return a.cleanupDueChatMedia(ctx, time.Now().Add(-time.Duration(days)*24*time.Hour)) }) if err != nil && !acquired { fail(w, http.StatusInternalServerError, 50001, "无法启动聊天媒体清理任务") return } if !acquired { fail(w, http.StatusConflict, 20001, "聊天媒体清理任务正在执行") return } if err != nil { fail(w, http.StatusBadGateway, 50001, "聊天媒体清理任务执行失败") return } a.recordChatMediaCleanupResult(r.Context(), result) a.audit(r, "cleanup_due", "chat_media", 0, map[string]any{"retentionDays": days, "result": result}) reply(w, result) } func (a *App) adminDeleteChatMedia(w http.ResponseWriter, r *http.Request) { messageID, err := pathID(r) if err != nil { fail(w, http.StatusBadRequest, 20001, "聊天媒体编号无效") return } result, acquired, runErr := a.withChatMediaCleanupLock(r.Context(), func(ctx context.Context) (chatMediaCleanupSummary, error) { summary := chatMediaCleanupSummary{Scanned: 1} deleted, cleanupErr := a.cleanupChatMedia(ctx, messageID, "管理员手动清理") if cleanupErr != nil { summary.Failed = 1 return summary, cleanupErr } if deleted { summary.Deleted = 1 } return summary, nil }) if runErr != nil && !acquired { fail(w, http.StatusInternalServerError, 50001, "无法启动聊天媒体清理任务") return } if !acquired { fail(w, http.StatusConflict, 20001, "聊天媒体清理任务正在执行") return } if runErr != nil { a.audit(r, "cleanup_failed", "chat_media", messageID, map[string]any{"error": chatMediaCleanupError(runErr)}) fail(w, http.StatusBadGateway, 50001, "文件清理失败,记录已保留并可重试") return } a.audit(r, "cleanup", "chat_media", messageID, map[string]any{"result": result}) reply(w, result) } func (a *App) withChatMediaCleanupLock(ctx context.Context, run func(context.Context) (chatMediaCleanupSummary, error)) (chatMediaCleanupSummary, bool, error) { conn, err := a.db.Conn(ctx) if err != nil { return chatMediaCleanupSummary{}, false, err } defer conn.Close() var acquired int if err = conn.QueryRowContext(ctx, `SELECT GET_LOCK(?,0)`, chatMediaCleanupLock).Scan(&acquired); err != nil || acquired != 1 { return chatMediaCleanupSummary{}, false, err } defer func() { releaseCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() _, _ = conn.ExecContext(releaseCtx, `SELECT RELEASE_LOCK(?)`, chatMediaCleanupLock) }() result, err := run(ctx) return result, true, err } func (a *App) cleanupDueChatMedia(ctx context.Context, before time.Time) (chatMediaCleanupSummary, error) { result := chatMediaCleanupSummary{} a.recoverStuckChatMediaCleanups(ctx) for batch := 0; batch < 50; batch++ { retryBefore := time.Now().Add(-time.Hour) rows, err := a.db.QueryContext(ctx, `SELECT message_id FROM im_message_media WHERE created_at0`, target.MediaAssetID.Int64) if err != nil { return err } for rows.Next() { var event chatMediaEvent if err = rows.Scan(&event.MessageID, &event.ConversationID, &event.Seq); err != nil { _ = rows.Close() return err } events = append(events, event) } if err = rows.Close(); err != nil { return err } if _, err = tx.ExecContext(ctx, `UPDATE media_assets SET status=0,deleted_at=COALESCE(deleted_at,NOW(3)) WHERE id=?`, target.MediaAssetID.Int64); err != nil { return err } if _, err = tx.ExecContext(ctx, `UPDATE im_message_media SET status=0,delete_reason=?,cleanup_error='',deleted_at=COALESCE(deleted_at,NOW(3)) WHERE media_asset_id=? AND status<>0`, reason, target.MediaAssetID.Int64); err != nil { return err } if err = tx.Commit(); err != nil { return err } for _, event := range events { a.broadcastChatMediaDeleted(ctx, event) } return nil } func (a *App) broadcastChatMediaDeleted(ctx context.Context, event chatMediaEvent) { rows, err := a.db.QueryContext(ctx, `SELECT user_id FROM im_conversation_members WHERE conversation_id=? AND status=1`, event.ConversationID) if err != nil { return } members := []int64{} for rows.Next() { var id int64 if rows.Scan(&id) == nil { members = append(members, id) } } _ = rows.Close() a.hub.broadcast(members, map[string]any{"command": "MESSAGE_MEDIA_DELETED", "data": map[string]any{"id": event.MessageID, "conversationId": event.ConversationID, "seq": event.Seq, "mediaDeleted": true}}) } func (a *App) recordChatMediaCleanupResult(ctx context.Context, result chatMediaCleanupSummary) { lastAt := time.Now().Format(time.RFC3339) lastResult := fmt.Sprintf("检查 %d 个,清理 %d 个,失败 %d 个", result.Scanned, result.Deleted, result.Failed) _, _ = a.db.ExecContext(ctx, `UPDATE system_configs SET config_value=? WHERE config_key='im.chat_media_cleanup_last_at'`, lastAt) _, _ = a.db.ExecContext(ctx, `UPDATE system_configs SET config_value=? WHERE config_key='im.chat_media_cleanup_last_result'`, lastResult) } func (a *App) runChatMediaCleanupWorker(ctx context.Context) { run := func() { a.recoverStuckChatMediaCleanups(ctx) if !a.configBool(ctx, "im.chat_media_retention_enabled", false) { return } workerCtx, cancel := context.WithTimeout(ctx, 20*time.Minute) defer cancel() days := clampChatMediaRetentionDays(a.configInt(workerCtx, "im.chat_media_retention_days", 90)) result, acquired, err := a.withChatMediaCleanupLock(workerCtx, func(runCtx context.Context) (chatMediaCleanupSummary, error) { return a.cleanupDueChatMedia(runCtx, time.Now().Add(-time.Duration(days)*24*time.Hour)) }) if err != nil && !acquired { log.Printf("chat media cleanup lock failed: %v", err) return } if !acquired { return } if err != nil { log.Printf("chat media cleanup failed: %v", err) return } a.recordChatMediaCleanupResult(workerCtx, result) } run() ticker := time.NewTicker(time.Hour) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-ticker.C: run() } } }