42 lines
1.4 KiB
Go
42 lines
1.4 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/tinode/chat/server/store"
|
|
"github.com/tinode/chat/server/store/types"
|
|
)
|
|
|
|
func replayKafkaDeadLetter(ctx context.Context, tenantID types.TenantID, deadID string, actor types.Uid, note string) (*types.KafkaDeadLetter, error) {
|
|
now := time.Now().UTC().Round(time.Millisecond)
|
|
dead, err := store.HTTPInfrastructure.MarkKafkaDeadLetterReplay(tenantID, deadID, actor.UserId(), note, now)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var envelope kafkaEnvelope
|
|
if err = json.Unmarshal(dead.Envelope, &envelope); err != nil {
|
|
return nil, err
|
|
}
|
|
if envelope.EventID == "" {
|
|
envelope.EventID = dead.EventID
|
|
}
|
|
if envelope.ParentEventID == "" {
|
|
envelope.ParentEventID = dead.ParentEventID
|
|
}
|
|
if err = publishKafkaEnvelope(ctx, strings.TrimSpace(dead.SourceTopic), envelope.EventID, envelope); err != nil {
|
|
return nil, err
|
|
}
|
|
if err = store.HTTPInfrastructure.MarkKafkaDeadLetter(tenantID, deadID, types.KafkaDeadStatusReplayed, actor.UserId(), note, now); err != nil {
|
|
return nil, err
|
|
}
|
|
return store.HTTPInfrastructure.GetKafkaDeadLetter(tenantID, deadID)
|
|
}
|
|
|
|
func resolveKafkaDeadLetter(tenantID types.TenantID, deadID, status string, actor types.Uid, note string) error {
|
|
now := time.Now().UTC().Round(time.Millisecond)
|
|
return store.HTTPInfrastructure.MarkKafkaDeadLetter(tenantID, deadID, status, actor.UserId(), note, now)
|
|
}
|