feat: add workbench kafka migration support
This commit is contained in:
richjohnrambo
+13
-423
@@ -269,12 +269,6 @@ func (a *adapter) Open(jsonconfig json.RawMessage) error {
|
||||
|
||||
// Actually opening the network connection.
|
||||
err = a.db.Ping()
|
||||
databaseMissing := isMissingDb(err)
|
||||
if databaseMissing {
|
||||
// Ignore missing database here. If we are initializing the database
|
||||
// missing DB is OK.
|
||||
err = nil
|
||||
}
|
||||
if err == nil {
|
||||
if config.SqlTimeout > 0 {
|
||||
a.sqlTimeout = time.Duration(config.SqlTimeout) * time.Second
|
||||
@@ -611,435 +605,31 @@ func createTenantRegistry(tx *sql.Tx, registryDB, businessDB string) error {
|
||||
return err
|
||||
}
|
||||
|
||||
// CreateDb initializes the storage.
|
||||
// CreateDb is intentionally disabled for this deployment.
|
||||
//
|
||||
// Database creation, table creation, schema migration, indexes and seed data
|
||||
// are handled by reviewed SQL files and the release process. The service and
|
||||
// local database tools must not create or mutate schema implicitly.
|
||||
func (a *adapter) CreateDb(reset bool) error {
|
||||
var err error
|
||||
var tx *sql.Tx
|
||||
|
||||
// Can't use an existing connection because it's configured with a database name which may not exist.
|
||||
// Don't care if it does not close cleanly.
|
||||
a.db.Close()
|
||||
|
||||
// This DSN has been parsed before and produced no error, not checking for errors here.
|
||||
cfg, _ := ms.ParseDSN(a.dsn)
|
||||
// Clear database name
|
||||
cfg.DBName = ""
|
||||
|
||||
a.db, err = sqlx.Open(instrumentedMySQLDriverName, cfg.FormatDSN())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if tx, err = a.db.Begin(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
defer func() {
|
||||
if err != nil {
|
||||
// FIXME: This is useless: MySQL auto-commits on every CREATE TABLE.
|
||||
// Maybe DROP DATABASE instead.
|
||||
tx.Rollback()
|
||||
}
|
||||
}()
|
||||
|
||||
registryDB := a.tenantRegistryDB
|
||||
if registryDB == "" {
|
||||
registryDB = "hotgo"
|
||||
}
|
||||
businessDB := a.dbName
|
||||
if businessDB == "" {
|
||||
businessDB = defaultDatabase
|
||||
}
|
||||
if reset {
|
||||
if _, err = tx.Exec("DROP DATABASE IF EXISTS " + quoteIdent(businessDB)); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err = tx.Exec("CREATE DATABASE IF NOT EXISTS " + quoteIdent(registryDB) + " CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci"); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err = tx.Exec("DROP TABLE IF EXISTS " + quoteIdent(registryDB) + ".im_tenant_config"); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err = tx.Exec("DROP TABLE IF EXISTS " + quoteIdent(registryDB) + ".im_tenant"); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
if err = createTenantRegistry(tx, registryDB, businessDB); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if _, err = tx.Exec("CREATE DATABASE IF NOT EXISTS " + quoteIdent(businessDB) + " CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci"); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if _, err = tx.Exec("USE " + quoteIdent(businessDB)); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if _, err = tx.Exec(
|
||||
`CREATE TABLE im_users(
|
||||
id BIGINT NOT NULL,
|
||||
account_no INT UNSIGNED NOT NULL DEFAULT 0,
|
||||
username VARCHAR(32) NOT NULL,
|
||||
createdat DATETIME(3) NOT NULL,
|
||||
updatedat DATETIME(3) NOT NULL,
|
||||
state SMALLINT NOT NULL DEFAULT 0,
|
||||
stateat DATETIME(3),
|
||||
access JSON,
|
||||
lastseen DATETIME,
|
||||
useragent VARCHAR(255) DEFAULT '',
|
||||
phone VARCHAR(32),
|
||||
public JSON,
|
||||
trusted JSON,
|
||||
PRIMARY KEY(id),
|
||||
UNIQUE INDEX uk_im_users_account_no(account_no),
|
||||
UNIQUE INDEX users_username(username),
|
||||
UNIQUE INDEX users_phone(phone),
|
||||
INDEX users_state_stateat(state, stateat),
|
||||
INDEX users_lastseen_updatedat(lastseen, updatedat)
|
||||
)`); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if _, err = tx.Exec(
|
||||
`CREATE TABLE im_user_security(
|
||||
id BIGINT NOT NULL,
|
||||
user_id BIGINT NOT NULL,
|
||||
auth_version INT NOT NULL DEFAULT 1,
|
||||
password_updated_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_user_security_user(user_id)
|
||||
)`); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err = createIMSecurityTables(tx); err != nil {
|
||||
return err
|
||||
}
|
||||
if err = createIMNotificationTables(tx); err != nil {
|
||||
return err
|
||||
}
|
||||
if err = ensureIMRegistrationInviteUseTable(tx); err != nil {
|
||||
return err
|
||||
}
|
||||
if a.registryDB == nil {
|
||||
if err = ensureIMRegistrationInviteControlTables(tx); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// Indexed devices. Normalized into a separate table.
|
||||
if _, err = tx.Exec(
|
||||
`CREATE TABLE im_devices(
|
||||
id INT NOT NULL AUTO_INCREMENT,
|
||||
userid BIGINT NOT NULL,
|
||||
hash CHAR(16) NOT NULL,
|
||||
deviceid TEXT NOT NULL,
|
||||
platform VARCHAR(32),
|
||||
lastseen DATETIME NOT NULL,
|
||||
lang VARCHAR(8),
|
||||
PRIMARY KEY(id),
|
||||
FOREIGN KEY(userid) REFERENCES im_users(id),
|
||||
UNIQUE INDEX devices_hash(hash)
|
||||
)`); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Authentication records for the basic authentication scheme.
|
||||
if _, err = tx.Exec(
|
||||
`CREATE TABLE im_auth(
|
||||
id INT NOT NULL AUTO_INCREMENT,
|
||||
userid BIGINT NOT NULL,
|
||||
authlvl INT NOT NULL,
|
||||
secret VARCHAR(255) NOT NULL,
|
||||
expires DATETIME,
|
||||
PRIMARY KEY(id),
|
||||
FOREIGN KEY(userid) REFERENCES im_users(id),
|
||||
UNIQUE INDEX auth_userid(userid)
|
||||
)`); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Topics
|
||||
if _, err = tx.Exec(
|
||||
`CREATE TABLE im_topics(
|
||||
id INT NOT NULL AUTO_INCREMENT,
|
||||
createdat DATETIME(3) NOT NULL,
|
||||
updatedat DATETIME(3) NOT NULL,
|
||||
state SMALLINT NOT NULL DEFAULT 0,
|
||||
stateat DATETIME(3),
|
||||
touchedat DATETIME(3),
|
||||
name CHAR(25) NOT NULL,
|
||||
usebt TINYINT DEFAULT 0,
|
||||
owner BIGINT NOT NULL DEFAULT 0,
|
||||
access JSON,
|
||||
seqid BIGINT NOT NULL DEFAULT 0,
|
||||
delid INT DEFAULT 0,
|
||||
subcnt INT DEFAULT 0,
|
||||
public JSON,
|
||||
trusted JSON,
|
||||
tags JSON,
|
||||
aux JSON,
|
||||
PRIMARY KEY(id),
|
||||
UNIQUE INDEX topics_name(name),
|
||||
INDEX topics_owner(owner),
|
||||
INDEX topics_state_stateat(state, stateat),
|
||||
INDEX topics_name_state_seqid(name, state, seqid)
|
||||
)`); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Create system topic 'sys'.
|
||||
if err = createSystemTopic(tx); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Indexed topic tags.
|
||||
if _, err = tx.Exec(
|
||||
`CREATE TABLE im_topictags(
|
||||
id INT NOT NULL AUTO_INCREMENT,
|
||||
topic CHAR(25) NOT NULL,
|
||||
tag VARCHAR(96) NOT NULL,
|
||||
PRIMARY KEY(id),
|
||||
FOREIGN KEY(topic) REFERENCES im_topics(name),
|
||||
INDEX topictags_tag(tag),
|
||||
UNIQUE INDEX topictags_topic_tag(topic, tag)
|
||||
)`); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Subscriptions
|
||||
if _, err = tx.Exec(
|
||||
`CREATE TABLE im_subscriptions(
|
||||
id INT NOT NULL AUTO_INCREMENT,
|
||||
createdat DATETIME(3) NOT NULL,
|
||||
updatedat DATETIME(3) NOT NULL,
|
||||
deletedat DATETIME(3),
|
||||
userid BIGINT NOT NULL,
|
||||
topic CHAR(25) NOT NULL,
|
||||
delid INT DEFAULT 0,
|
||||
recvseqid BIGINT DEFAULT 0,
|
||||
readseqid BIGINT DEFAULT 0,
|
||||
modewant CHAR(8),
|
||||
modegiven CHAR(8),
|
||||
private JSON,
|
||||
PRIMARY KEY(id),
|
||||
FOREIGN KEY(userid) REFERENCES im_users(id),
|
||||
UNIQUE INDEX subscriptions_topic_userid(topic, userid),
|
||||
INDEX subscriptions_topic(topic),
|
||||
INDEX subscriptions_deletedat(deletedat),
|
||||
INDEX subscriptions_userid_topic_deletedat(userid, topic, deletedat)
|
||||
)`); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Messages
|
||||
if _, err = tx.Exec(
|
||||
`CREATE TABLE im_messages(
|
||||
id INT NOT NULL AUTO_INCREMENT,
|
||||
createdat DATETIME(3) NOT NULL,
|
||||
updatedat DATETIME(3) NOT NULL,
|
||||
deletedat DATETIME(3),
|
||||
delid INT DEFAULT 0,
|
||||
seqid BIGINT NOT NULL,
|
||||
topic CHAR(25) NOT NULL,` +
|
||||
"`from` BIGINT NOT NULL," +
|
||||
`client_msg_id VARCHAR(64),
|
||||
version INT NOT NULL DEFAULT 1,
|
||||
recalled_at DATETIME(3),
|
||||
recalled_by BIGINT,
|
||||
head JSON,
|
||||
content JSON,
|
||||
PRIMARY KEY(id),
|
||||
FOREIGN KEY(topic) REFERENCES im_topics(name),
|
||||
UNIQUE INDEX messages_sender_client_mid(` + "`from`" + `, client_msg_id),
|
||||
UNIQUE INDEX messages_topic_seqid(topic, seqid)
|
||||
)`); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if _, err = tx.Exec(
|
||||
`CREATE TABLE im_message_revisions(
|
||||
id BIGINT NOT NULL AUTO_INCREMENT,
|
||||
createdat DATETIME(3) NOT NULL,
|
||||
updatedat DATETIME(3) NOT NULL,
|
||||
topic CHAR(25) NOT NULL,
|
||||
seqid BIGINT NOT NULL,
|
||||
version INT NOT NULL,
|
||||
actor BIGINT NOT NULL,
|
||||
head JSON,
|
||||
content JSON,
|
||||
reason VARCHAR(256) NOT NULL DEFAULT '',
|
||||
PRIMARY KEY(id),
|
||||
UNIQUE INDEX uk_im_message_revisions_version(topic, seqid, version)
|
||||
)`); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err = createIMRelationTables(tx); err != nil {
|
||||
return err
|
||||
}
|
||||
if err = createIMCallTables(tx); err != nil {
|
||||
return err
|
||||
}
|
||||
if err = createIMWorkbenchTables(tx); err != nil {
|
||||
return err
|
||||
}
|
||||
if err = createIMOrganizationTables(tx); err != nil {
|
||||
return err
|
||||
}
|
||||
if err = createIMAttendanceTables(tx); err != nil {
|
||||
return err
|
||||
}
|
||||
if err = createIMMeetingTables(tx); err != nil {
|
||||
return err
|
||||
}
|
||||
if err = createIMReportTables(tx); err != nil {
|
||||
return err
|
||||
}
|
||||
if err = createIMMailboxTables(tx); err != nil {
|
||||
return err
|
||||
}
|
||||
if err = createIMUserProductTables(tx); err != nil {
|
||||
return err
|
||||
}
|
||||
if err = ensureIMGroupBusinessTables(tx); err != nil {
|
||||
return err
|
||||
}
|
||||
if err = ensureIMHTTPInfrastructureTables(tx); err != nil {
|
||||
return err
|
||||
}
|
||||
if err = ensureIMAdminGovernanceTables(tx); err != nil {
|
||||
return err
|
||||
}
|
||||
if err = ensureIMFileReviewTables(tx); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Deletion log
|
||||
if _, err = tx.Exec(
|
||||
`CREATE TABLE im_dellog(
|
||||
id INT NOT NULL AUTO_INCREMENT,
|
||||
topic CHAR(25) NOT NULL,
|
||||
deletedfor BIGINT NOT NULL DEFAULT 0,
|
||||
delid INT NOT NULL,
|
||||
low INT NOT NULL,
|
||||
hi INT NOT NULL,
|
||||
PRIMARY KEY(id),
|
||||
FOREIGN KEY(topic) REFERENCES im_topics(name),
|
||||
INDEX dellog_topic_delid_deletedfor(topic,delid,deletedfor),
|
||||
INDEX dellog_topic_deletedfor_low_hi(topic,deletedfor,low,hi),
|
||||
INDEX dellog_deletedfor(deletedfor)
|
||||
)`); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// User credentials
|
||||
if _, err = tx.Exec(
|
||||
`CREATE TABLE im_credentials(
|
||||
id INT NOT NULL AUTO_INCREMENT,
|
||||
createdat DATETIME(3) NOT NULL,
|
||||
updatedat DATETIME(3) NOT NULL,
|
||||
deletedat DATETIME(3),
|
||||
method VARCHAR(16) NOT NULL,
|
||||
value VARCHAR(128) NOT NULL,
|
||||
synthetic VARCHAR(192) NOT NULL,
|
||||
userid BIGINT NOT NULL,
|
||||
resp VARCHAR(255),
|
||||
done TINYINT NOT NULL DEFAULT 0,
|
||||
retries INT NOT NULL DEFAULT 0,
|
||||
PRIMARY KEY(id),
|
||||
FOREIGN KEY(userid) REFERENCES im_users(id),
|
||||
UNIQUE credentials_uniqueness(synthetic)
|
||||
)`); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Records of uploaded files.
|
||||
// Don't add FOREIGN KEY on userid. It's not needed and it will break user deletion.
|
||||
if _, err = tx.Exec(
|
||||
`CREATE TABLE im_fileuploads(
|
||||
id BIGINT NOT NULL,
|
||||
createdat DATETIME(3) NOT NULL,
|
||||
updatedat DATETIME(3) NOT NULL,
|
||||
userid BIGINT,
|
||||
status INT NOT NULL,
|
||||
mimetype VARCHAR(255) NOT NULL,
|
||||
size BIGINT NOT NULL,
|
||||
etag VARCHAR(128),
|
||||
location VARCHAR(2048) NOT NULL,
|
||||
PRIMARY KEY(id),
|
||||
INDEX fileuploads_status(status)
|
||||
)`); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Links between uploaded files and the topics, users or messages they are attached to.
|
||||
if _, err = tx.Exec(
|
||||
`CREATE TABLE im_filemsglinks(
|
||||
id INT NOT NULL AUTO_INCREMENT,
|
||||
createdat DATETIME(3) NOT NULL,
|
||||
fileid BIGINT NOT NULL,
|
||||
msgid INT,
|
||||
topic CHAR(25),
|
||||
userid BIGINT,
|
||||
PRIMARY KEY(id),
|
||||
FOREIGN KEY(fileid) REFERENCES im_fileuploads(id) ON DELETE CASCADE,
|
||||
FOREIGN KEY(msgid) REFERENCES im_messages(id) ON DELETE CASCADE,
|
||||
FOREIGN KEY(topic) REFERENCES im_topics(name) ON DELETE CASCADE,
|
||||
FOREIGN KEY(userid) REFERENCES im_users(id) ON DELETE CASCADE
|
||||
)`); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if _, err = tx.Exec(
|
||||
`CREATE TABLE im_kvmeta(` +
|
||||
"id BIGINT NOT NULL," +
|
||||
"`key` VARCHAR(64) NOT NULL," +
|
||||
"createdat DATETIME(3)," +
|
||||
"`value` TEXT," +
|
||||
"PRIMARY KEY(id)," +
|
||||
"UNIQUE INDEX uk_im_kvmeta_key(`key`)," +
|
||||
"INDEX kvmeta_createdat_key(createdat, `key`)" +
|
||||
`)`); err != nil {
|
||||
return err
|
||||
}
|
||||
versionID, err := nextSnowflakeID()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err = tx.Exec("INSERT INTO im_kvmeta(id, `key`, `value`) VALUES(?,'version',?)", versionID, adpVersion); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return tx.Commit()
|
||||
return errors.New("mysql schema auto-creation is disabled; apply reviewed SQL manually before starting the service")
|
||||
}
|
||||
|
||||
// UpgradeDb upgrades the database, if necessary.
|
||||
// UpgradeDb is intentionally schema-read-only.
|
||||
//
|
||||
// Automatic ALTER/CREATE upgrade steps are disabled. If the database version
|
||||
// differs from the adapter version, prepare and apply a reviewed SQL migration,
|
||||
// then update im_kvmeta.version as part of that migration.
|
||||
func (a *adapter) UpgradeDb() error {
|
||||
if _, err := a.GetDbVersion(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := a.ensureMessageTopicSeqIndex(); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := a.ensureTopicNameUniqueIndex(); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := a.ensureTopicSeqMatchesMessages(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if a.version == adpVersion {
|
||||
return nil
|
||||
}
|
||||
|
||||
return errors.New("mysql independent-database multitenancy requires a fresh schema; automatic upgrade from version " +
|
||||
strconv.Itoa(a.version) + " to " + strconv.Itoa(adpVersion) + " is not supported")
|
||||
return errors.New("mysql schema auto-upgrade is disabled; apply reviewed SQL manually to upgrade from version " +
|
||||
strconv.Itoa(a.version) + " to " + strconv.Itoa(adpVersion))
|
||||
}
|
||||
|
||||
func (a *adapter) ensureMessageTopicSeqIndex() error {
|
||||
|
||||
@@ -24,6 +24,7 @@ const (
|
||||
kafkaTopicMessagePersist = "im.message.persist"
|
||||
kafkaTopicGroupMessagePersist = "im.group.message.persist"
|
||||
kafkaTopicMessagePush = "im.message.push"
|
||||
kafkaTopicWorkbenchExternal = "im.workbench.event"
|
||||
kafkaTopicFileSecurity = "im.file.security"
|
||||
kafkaTopicCacheInvalidate = "im.cache.invalidate"
|
||||
kafkaTopicAuditEvent = "im.audit.event"
|
||||
@@ -210,6 +211,7 @@ func kafkaTopics(overrides map[string]string) map[string]string {
|
||||
"message_persist": kafkaTopicMessagePersist,
|
||||
"group_message_persist": kafkaTopicGroupMessagePersist,
|
||||
"message_push": kafkaTopicMessagePush,
|
||||
"workbench_external": kafkaTopicWorkbenchExternal,
|
||||
"file_security": kafkaTopicFileSecurity,
|
||||
"cache_invalidate": kafkaTopicCacheInvalidate,
|
||||
"audit_event": kafkaTopicAuditEvent,
|
||||
|
||||
@@ -0,0 +1,163 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/segmentio/kafka-go"
|
||||
"github.com/tinode/chat/server/logs"
|
||||
"github.com/tinode/chat/server/store"
|
||||
"github.com/tinode/chat/server/store/types"
|
||||
)
|
||||
|
||||
type kafkaExternalWorkbenchEvent struct {
|
||||
EventID string `json:"event_id"`
|
||||
TraceID string `json:"trace_id,omitempty"`
|
||||
TenantID types.TenantID `json:"tenant_id,omitempty"`
|
||||
TenantCode string `json:"tenant_code,omitempty"`
|
||||
ActorUserID int64 `json:"actor_user_id"`
|
||||
Module string `json:"module"`
|
||||
Action string `json:"action"`
|
||||
ResourceID string `json:"resource_id"`
|
||||
TargetUserIDs []int64 `json:"target_user_ids,omitempty"`
|
||||
Title string `json:"title,omitempty"`
|
||||
Summary string `json:"summary,omitempty"`
|
||||
OccurredAt time.Time `json:"occurred_at"`
|
||||
Data map[string]any `json:"data,omitempty"`
|
||||
}
|
||||
|
||||
func startKafkaWorkbenchEventConsumer() {
|
||||
startKafkaConsumer("workbench_external", handleKafkaWorkbenchEvent)
|
||||
}
|
||||
|
||||
func handleKafkaWorkbenchEvent(ctx context.Context, envelope kafkaEnvelope, msg kafka.Message) error {
|
||||
payload, normalized, err := decodeKafkaWorkbenchEvent(envelope, msg)
|
||||
if err != nil {
|
||||
logs.Warn.Printf("kafka workbench event malformed: topic=%s offset=%d err=%v", msg.Topic, msg.Offset, err)
|
||||
return err
|
||||
}
|
||||
tenantID, err := kafkaTenantIDFromEnvelope(normalized)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
consume, claimed, err := beginKafkaConsumeForTopic(tenantID, normalized, "workbench_external")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !claimed {
|
||||
logs.Info.Printf("kafka workbench event skipped: event=%s group=%s status=%s", normalized.EventID, kafkaBus.config.ConsumerGroup, consume.Status)
|
||||
return nil
|
||||
}
|
||||
if err = executeKafkaExternalWorkbenchEvent(tenantID, normalized, payload); err != nil {
|
||||
reason := kafkaCommandErrorReason(err)
|
||||
failed, markErr := store.HTTPInfrastructure.MarkKafkaConsumeFailed(tenantID, normalized.EventID, kafkaBus.config.ConsumerGroup, reason, err.Error(), time.Now().UTC().Round(time.Millisecond))
|
||||
if markErr != nil {
|
||||
return markErr
|
||||
}
|
||||
if failed != nil && failed.RetryCount >= kafkaBus.config.MaxRetryCount {
|
||||
return markKafkaDeadLetter(ctx, tenantID, normalized, failed, reason, err.Error())
|
||||
}
|
||||
if failed != nil {
|
||||
sleepKafkaRetryBackoff(failed.RetryCount)
|
||||
}
|
||||
return err
|
||||
}
|
||||
return store.HTTPInfrastructure.MarkKafkaConsumeSucceeded(tenantID, normalized.EventID, kafkaBus.config.ConsumerGroup, time.Now().UTC().Round(time.Millisecond))
|
||||
}
|
||||
|
||||
func decodeKafkaWorkbenchEvent(envelope kafkaEnvelope, msg kafka.Message) (kafkaExternalWorkbenchEvent, kafkaEnvelope, error) {
|
||||
var payload kafkaExternalWorkbenchEvent
|
||||
if len(envelope.Payload) > 0 {
|
||||
if err := json.Unmarshal(envelope.Payload, &payload); err != nil {
|
||||
return payload, envelope, err
|
||||
}
|
||||
} else if err := json.Unmarshal(msg.Value, &payload); err != nil {
|
||||
return payload, envelope, err
|
||||
}
|
||||
if strings.TrimSpace(payload.EventID) == "" {
|
||||
payload.EventID = envelope.EventID
|
||||
}
|
||||
if strings.TrimSpace(payload.Action) == "" && envelope.EventType != "" {
|
||||
payload.Action = strings.TrimPrefix(envelope.EventType, "workbench.")
|
||||
}
|
||||
if strings.TrimSpace(payload.EventID) == "" || strings.TrimSpace(payload.Action) == "" || strings.TrimSpace(payload.ResourceID) == "" {
|
||||
return payload, envelope, types.ErrMalformed
|
||||
}
|
||||
if payload.OccurredAt.IsZero() {
|
||||
payload.OccurredAt = time.Now().UTC().Round(time.Millisecond)
|
||||
}
|
||||
normalized := kafkaEnvelope{
|
||||
EventID: payload.EventID,
|
||||
TraceID: firstNonEmpty(payload.TraceID, envelope.TraceID),
|
||||
TenantCode: firstNonEmpty(payload.TenantCode, envelope.TenantCode),
|
||||
TenantID: firstTenantID(payload.TenantID, envelope.TenantID),
|
||||
SourceService: firstNonEmpty(envelope.SourceService, "im-workbench"),
|
||||
TargetService: firstNonEmpty(envelope.TargetService, "im-server"),
|
||||
EventType: "workbench." + payload.Action,
|
||||
BizType: firstNonEmpty(payload.Module, envelope.BizType, "workbench"),
|
||||
BizID: firstNonEmpty(payload.ResourceID, envelope.BizID),
|
||||
OccurredAt: payload.OccurredAt,
|
||||
Payload: msg.Value,
|
||||
}
|
||||
if payload.ActorUserID > 0 {
|
||||
normalized.Actor = &kafkaActor{Type: "user", ID: types.Uid(payload.ActorUserID).UserId()}
|
||||
}
|
||||
return payload, normalized, nil
|
||||
}
|
||||
|
||||
func executeKafkaExternalWorkbenchEvent(tenantID types.TenantID, envelope kafkaEnvelope, payload kafkaExternalWorkbenchEvent) error {
|
||||
if store.Security == nil {
|
||||
return types.ErrUnsupported
|
||||
}
|
||||
actor := types.Uid(payload.ActorUserID)
|
||||
data := types.KVMap{
|
||||
"module": payload.Module,
|
||||
"resource_id": payload.ResourceID,
|
||||
}
|
||||
for k, v := range payload.Data {
|
||||
data[k] = v
|
||||
}
|
||||
targets := compactWorkbenchTargets(payload.TargetUserIDs, actor)
|
||||
for _, target := range targets {
|
||||
targetID := target.UserId()
|
||||
eventID := stableIMEventID("workbench", envelope.EventID, targetID)
|
||||
if target != actor && payload.Title != "" {
|
||||
if err := createIMSystemNotification(tenantID, target, stableIMEventID(eventID, "notification"), payload.Action,
|
||||
payload.Title, payload.Summary, data); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if err := enqueueIMBusinessEvent(tenantID, eventID, "user", targetID, actor, payload.Action, data); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func compactWorkbenchTargets(raw []int64, actor types.Uid) []types.Uid {
|
||||
seen := make(map[types.Uid]bool, len(raw)+1)
|
||||
targets := make([]types.Uid, 0, len(raw)+1)
|
||||
for _, value := range raw {
|
||||
uid := types.Uid(value)
|
||||
if uid.IsZero() || seen[uid] {
|
||||
continue
|
||||
}
|
||||
seen[uid] = true
|
||||
targets = append(targets, uid)
|
||||
}
|
||||
if len(targets) == 0 && !actor.IsZero() {
|
||||
targets = append(targets, actor)
|
||||
}
|
||||
return targets
|
||||
}
|
||||
|
||||
func firstTenantID(values ...types.TenantID) types.TenantID {
|
||||
for _, value := range values {
|
||||
if !value.IsZero() {
|
||||
return value
|
||||
}
|
||||
}
|
||||
return types.ZeroTenantID
|
||||
}
|
||||
@@ -0,0 +1,89 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/segmentio/kafka-go"
|
||||
"github.com/tinode/chat/server/store"
|
||||
"github.com/tinode/chat/server/store/types"
|
||||
)
|
||||
|
||||
func TestKafkaTopicsIncludeExternalWorkbenchEvent(t *testing.T) {
|
||||
topics := kafkaTopics(nil)
|
||||
if topics["workbench_external"] != "im.workbench.event" {
|
||||
t.Fatalf("unexpected workbench external topic: %q", topics["workbench_external"])
|
||||
}
|
||||
if _, ok := topics["workbench_event"]; ok {
|
||||
t.Fatal("legacy in-process workbench_event topic key must not be configured")
|
||||
}
|
||||
}
|
||||
|
||||
func TestDecodeKafkaWorkbenchEventDirectPayload(t *testing.T) {
|
||||
msg := kafka.Message{Value: []byte(`{
|
||||
"event_id":"wbe_1",
|
||||
"trace_id":"trace-1",
|
||||
"tenant_id":7,
|
||||
"tenant_code":"default",
|
||||
"actor_user_id":42,
|
||||
"module":"meeting",
|
||||
"action":"meeting.created",
|
||||
"resource_id":"meeting-1",
|
||||
"target_user_ids":[42,43],
|
||||
"title":"会议通知",
|
||||
"summary":"周会",
|
||||
"occurred_at":"2026-09-13T10:00:00Z",
|
||||
"data":{"meeting_id":"meeting-1"}
|
||||
}`)}
|
||||
|
||||
payload, envelope, err := decodeKafkaWorkbenchEvent(kafkaEnvelope{}, msg)
|
||||
if err != nil {
|
||||
t.Fatalf("decodeKafkaWorkbenchEvent returned error: %v", err)
|
||||
}
|
||||
if payload.EventID != "wbe_1" || payload.ActorUserID != 42 || len(payload.TargetUserIDs) != 2 {
|
||||
t.Fatalf("unexpected payload: %#v", payload)
|
||||
}
|
||||
if envelope.EventID != "wbe_1" || envelope.EventType != "workbench.meeting.created" ||
|
||||
envelope.BizType != "meeting" || envelope.BizID != "meeting-1" {
|
||||
t.Fatalf("unexpected normalized envelope: %#v", envelope)
|
||||
}
|
||||
if envelope.Actor == nil || envelope.Actor.ID != types.Uid(42).UserId() {
|
||||
t.Fatalf("unexpected actor: %#v", envelope.Actor)
|
||||
}
|
||||
if envelope.OccurredAt.IsZero() || envelope.TenantID != types.TenantID(7) {
|
||||
t.Fatalf("unexpected occurrence or tenant: %#v", envelope)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExecuteKafkaExternalWorkbenchEventCreatesNotificationForNonActor(t *testing.T) {
|
||||
oldSecurity := store.Security
|
||||
defer func() { store.Security = oldSecurity }()
|
||||
|
||||
security := &testSecurityStore{}
|
||||
store.Security = security
|
||||
|
||||
payload := kafkaExternalWorkbenchEvent{
|
||||
EventID: "wbe_1",
|
||||
TenantID: 7,
|
||||
ActorUserID: 42,
|
||||
Module: "meeting",
|
||||
Action: "meeting.created",
|
||||
ResourceID: "meeting-1",
|
||||
TargetUserIDs: []int64{42, 43},
|
||||
Title: "会议通知",
|
||||
Summary: "周会",
|
||||
OccurredAt: time.Date(2026, 9, 13, 10, 0, 0, 0, time.UTC),
|
||||
Data: map[string]any{"meeting_id": "meeting-1"},
|
||||
}
|
||||
envelope := kafkaEnvelope{EventID: payload.EventID}
|
||||
|
||||
if err := executeKafkaExternalWorkbenchEvent(7, envelope, payload); err != nil {
|
||||
t.Fatalf("executeKafkaExternalWorkbenchEvent returned error: %v", err)
|
||||
}
|
||||
if len(security.notifications) != 1 {
|
||||
t.Fatalf("expected one notification for non-actor target, got %#v", security.notifications)
|
||||
}
|
||||
if security.notifications[0].UserID != types.Uid(43) || security.notifications[0].EventType != "meeting.created" {
|
||||
t.Fatalf("unexpected notification: %#v", security.notifications[0])
|
||||
}
|
||||
}
|
||||
@@ -750,6 +750,7 @@ func main() {
|
||||
startKafkaConfigEventConsumer()
|
||||
startKafkaCacheInvalidateConsumer()
|
||||
startKafkaMessagePersistConsumer()
|
||||
startKafkaWorkbenchEventConsumer()
|
||||
startKafkaDeadLetterConsumer()
|
||||
|
||||
// Start accepting cluster traffic.
|
||||
|
||||
@@ -137,6 +137,7 @@
|
||||
"message_persist": "im.message.persist",
|
||||
"group_message_persist": "im.group.message.persist",
|
||||
"message_push": "im.message.push",
|
||||
"workbench_external": "im.workbench.event",
|
||||
"file_security": "im.file.security",
|
||||
"cache_invalidate": "im.cache.invalidate",
|
||||
"audit_event": "im.audit.event",
|
||||
|
||||
@@ -137,6 +137,7 @@
|
||||
"message_persist": "im.message.persist",
|
||||
"group_message_persist": "im.group.message.persist",
|
||||
"message_push": "im.message.push",
|
||||
"workbench_external": "im.workbench.event",
|
||||
"file_security": "im.file.security",
|
||||
"cache_invalidate": "im.cache.invalidate",
|
||||
"audit_event": "im.audit.event",
|
||||
|
||||
Reference in New Issue
Block a user