pro/im_server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add file safety config integration

Browse Source
This commit is contained in:
richjohnrambo
2026-08-23 23:50:21 +08:00
parent 7b942da975
commit ed9413d923
20 changed files with 1407 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/db/mysql/adapter.go
+15 -1
View File
@@ -255,6 +255,9 @@ func (a *adapter) Open(jsonconfig json.RawMessage) error {
if err = ensureIMHTTPInfrastructureTables(a.db); err != nil {
return errors.New("mysql adapter failed to ensure HTTP infrastructure schema: " + err.Error())
}
if err = ensureIMFileReviewTables(a.db); err != nil {
return errors.New("mysql adapter failed to ensure file review schema: " + err.Error())
}
if err = ensureSnowflakePrimaryKeys(a.db); err != nil {
return errors.New("mysql adapter failed to ensure snowflake primary keys: " + err.Error())
}
@@ -458,6 +461,10 @@ func (a *adapter) TenantAdapterFor(tenantID t.TenantID) (dbadapter.TenantBusines
db.Close()
return nil, nil, err
}
if err = ensureIMFileReviewTables(db); err != nil {
db.Close()
return nil, nil, err
}
if err = ensureSnowflakePrimaryKeys(db); err != nil {
db.Close()
return nil, nil, err
@@ -877,6 +884,9 @@ func (a *adapter) CreateDb(reset bool) error {
if err = ensureIMHTTPInfrastructureTables(tx); err != nil {
return err
}
if err = ensureIMFileReviewTables(tx); err != nil {
return err
}
// Deletion log
if _, err = tx.Exec(
@@ -4465,9 +4475,13 @@ func (a *adapter) FileFinishUpload(fd *t.FileDef, success bool, size int64) (*t.
if err != nil {
return nil, err
}
fd.UpdatedAt = now
fd.Size = size
if err = insertFileReviewTasks(ctx, tx, fd); err != nil {
return nil, err
}
fd.Status = t.UploadCompleted
fd.Size = size
} else {
// Deleting the record: there is no value in keeping it in the DB.
_, err = tx.ExecContext(ctx, "DELETE FROM im_fileuploads WHERE id=?", store.DecodeUid(fd.Uid()))
server/db/mysql/file_review_schema.go
+37
View File
@@ -0,0 +1,37 @@
//go:build mysql
// +build mysql
package mysql
// ensureIMFileReviewTables installs async file safety review state in the
// currently selected tenant database. Tenant databases intentionally do not
// carry tenant_id columns.
func ensureIMFileReviewTables(exec sqlExecutor) error {
statements := []string{
`CREATE TABLE IF NOT EXISTS im_file_review_tasks (
id BIGINT NOT NULL, task_id VARCHAR(64) NOT NULL, file_id BIGINT NOT NULL,
user_id BIGINT NOT NULL DEFAULT 0, task_type VARCHAR(32) NOT NULL,
provider VARCHAR(64) NOT NULL, storage_provider VARCHAR(32) NOT NULL DEFAULT '',
object_name VARCHAR(2048) NOT NULL DEFAULT '', mime_type VARCHAR(255) NOT NULL DEFAULT '',
file_size BIGINT NOT NULL DEFAULT 0, provider_task_id VARCHAR(128) NOT NULL DEFAULT '',
status VARCHAR(16) NOT NULL DEFAULT 'pending', risk_level VARCHAR(32) NOT NULL DEFAULT '',
risk_labels JSON, result_json JSON, error_code VARCHAR(64) NOT NULL DEFAULT '',
error_message VARCHAR(512) NOT NULL DEFAULT '', attempts INT NOT NULL DEFAULT 0,
next_retry_at DATETIME(3), completed_at DATETIME(3),
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
PRIMARY KEY(id), UNIQUE INDEX uk_im_file_review_tasks_task(task_id),
UNIQUE INDEX uk_im_file_review_tasks_file_provider(file_id, task_type, provider),
INDEX idx_im_file_review_tasks_pending(status, next_retry_at, id),
INDEX idx_im_file_review_tasks_file(file_id),
INDEX idx_im_file_review_tasks_object(storage_provider, object_name(191)),
INDEX idx_im_file_review_tasks_provider_task(provider, provider_task_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
}
for _, statement := range statements {
if _, err := exec.Exec(statement); err != nil {
return err
}
}
return nil
}
server/db/mysql/file_review_schema_test.go
+32
View File
@@ -0,0 +1,32 @@
//go:build mysql
package mysql
import (
"strings"
"testing"
)
func TestEnsureIMFileReviewTablesUsesSelectedTenantDatabase(t *testing.T) {
exec := &recordingGroupSchemaExecutor{}
if err := ensureIMFileReviewTables(exec); err != nil {
t.Fatal(err)
}
if len(exec.statements) != 1 {
t.Fatalf("created %d file review tables, want 1", len(exec.statements))
}
joined := strings.ToLower(strings.Join(exec.statements, "\n"))
for _, expected := range []string{
"im_file_review_tasks",
"unique index uk_im_file_review_tasks_file_provider(file_id, task_type, provider)",
"index idx_im_file_review_tasks_pending(status, next_retry_at, id)",
"index idx_im_file_review_tasks_object(storage_provider, object_name(191))",
} {
if !strings.Contains(joined, expected) {
t.Fatalf("file review schema is missing %q", expected)
}
}
if strings.Contains(joined, "tinode.") || strings.Contains(joined, "create database") || strings.Contains(joined, "tenant_id") {
t.Fatalf("file review schema must stay inside the selected tenant database: %s", joined)
}
}
server/db/mysql/file_review_tasks.go
+47
View File
@@ -0,0 +1,47 @@
//go:build mysql
// +build mysql
package mysql
import (
"context"
"database/sql"
"github.com/tinode/chat/server/store"
t "github.com/tinode/chat/server/store/types"
)
type fileReviewExecContext interface {
ExecContext(context.Context, string, ...any) (sql.Result, error)
}
func insertFileReviewTasks(ctx context.Context, exec fileReviewExecContext, fd *t.FileDef) error {
fileID := decodeUID(fd.Uid())
userID := int64(0)
if fd.User != "" {
userID = decodeUID(t.ParseUid(fd.User))
}
tasks := []struct {
taskType string
provider string
}{
{taskType: t.FileReviewTaskTypeContent, provider: t.FileReviewProviderAliyunGreen},
{taskType: t.FileReviewTaskTypeMalware, provider: t.FileReviewProviderAliyunSAS},
}
for _, task := range tasks {
id, err := nextSnowflakeID()
if err != nil {
return err
}
taskID := store.EncodeUid(id).String()
_, err = exec.ExecContext(ctx, `INSERT IGNORE INTO im_file_review_tasks
(id,task_id,file_id,user_id,task_type,provider,storage_provider,object_name,mime_type,file_size,status,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)`,
id, taskID, fileID, userID, task.taskType, task.provider, "", fd.Location, fd.MimeType, fd.Size,
t.FileReviewStatusPending, fd.UpdatedAt, fd.UpdatedAt)
if err != nil {
return err
}
}
return nil
}
server/db/mysql/schema.sql
+1
View File
@@ -12,6 +12,7 @@
# - Default tenant business database: tinode
# - im_users, im_topics, im_subscriptions, im_messages and extension tables
# - im_http_idempotency and im_business_events_outbox for reliable HTTP writes
# - im_file_review_tasks for asynchronous content safety and malware checks
# - business tables do not store tenant_id in independent-database mode
#
# The previous shared-table tenant_id draft has been removed from this file to
server/db/mysql/tenant_adapter.go
+5 -1
View File
@@ -5439,8 +5439,12 @@ func (a *adapter) TenantFileFinishUpload(id t.TenantID, fd *t.FileDef, success b
if success {
_, err = tx.Exec("UPDATE im_fileuploads SET updatedat=?,status=?,size=?,etag=?,location=? WHERE tenant_id=? AND id=?",
now, t.UploadCompleted, size, fd.ETag, fd.Location, id, decodeUID(fd.Uid()))
if err == nil {
fd.UpdatedAt = now
fd.Size = size
err = insertFileReviewTasks(ctx, tx, fd)
}
fd.Status = t.UploadCompleted
fd.Size = size
} else {
_, err = tx.Exec("DELETE FROM im_fileuploads WHERE tenant_id=? AND id=?", id, decodeUID(fd.Uid()))
fd.Status = t.UploadFailed
server/db/mysql/third_party_config.go
+70
View File
@@ -0,0 +1,70 @@
//go:build mysql
// +build mysql
package mysql
import (
"database/sql"
"strings"
"github.com/jmoiron/sqlx"
t "github.com/tinode/chat/server/store/types"
)
func (a *adapter) ThirdPartyConfig(group string, keys ...string) (map[string]string, error) {
db := a.registryDB
if db == nil {
db = a.db
}
if db == nil {
return nil, t.ErrInternal
}
group = strings.TrimSpace(group)
if group == "" {
return nil, t.ErrMalformed
}
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
rows := []struct {
Key string `db:"key"`
Value string `db:"value"`
}{}
var err error
if len(keys) == 0 {
err = db.SelectContext(ctx, &rows, `SELECT `+"`key`"+`, value FROM hg_sys_config
WHERE `+"`group`"+`=? AND status=1`, group)
} else {
normalized := make([]string, 0, len(keys))
for _, key := range keys {
if key = strings.TrimSpace(key); key != "" {
normalized = append(normalized, key)
}
}
if len(normalized) == 0 {
return map[string]string{}, nil
}
query, args, inErr := sqlx.In(`SELECT `+"`key`"+`, value FROM hg_sys_config
WHERE `+"`group`"+`=? AND status=1 AND `+"`key`"+` IN (?)`, group, normalized)
if inErr != nil {
return nil, inErr
}
query = db.Rebind(query)
err = db.SelectContext(ctx, &rows, query, args...)
}
if err != nil {
if err == sql.ErrNoRows {
return map[string]string{}, nil
}
return nil, err
}
result := make(map[string]string, len(rows))
for _, row := range rows {
result[row.Key] = row.Value
}
return result, nil
}