127 lines
5.2 KiB
Go
127 lines
5.2 KiB
Go
package store
|
|
|
|
import (
|
|
"time"
|
|
|
|
adapter "github.com/tinode/chat/server/db"
|
|
"github.com/tinode/chat/server/store/types"
|
|
)
|
|
|
|
type HTTPInfrastructurePersistenceInterface interface {
|
|
BeginIdempotency(types.TenantID, *types.HTTPIdempotencyRecord) (*types.HTTPIdempotencyRecord, bool, error)
|
|
CompleteIdempotency(types.TenantID, *types.HTTPIdempotencyRecord) error
|
|
AbortIdempotency(types.TenantID, *types.HTTPIdempotencyRecord) error
|
|
BeginKafkaConsume(types.TenantID, *types.KafkaConsumeRecord, time.Time) (*types.KafkaConsumeRecord, bool, error)
|
|
MarkKafkaConsumeSucceeded(types.TenantID, string, string, time.Time) error
|
|
MarkKafkaConsumeFailed(types.TenantID, string, string, string, string, time.Time) (*types.KafkaConsumeRecord, error)
|
|
MarkKafkaConsumeDead(types.TenantID, *types.KafkaDeadLetter, time.Time) error
|
|
ListKafkaDeadLetters(types.TenantID, types.KafkaDeadLetterListOptions) (*types.KafkaDeadLetterPage, error)
|
|
GetKafkaDeadLetter(types.TenantID, string) (*types.KafkaDeadLetter, error)
|
|
MarkKafkaDeadLetter(types.TenantID, string, string, string, string, time.Time) error
|
|
MarkKafkaDeadLetterReplay(types.TenantID, string, string, string, time.Time) (*types.KafkaDeadLetter, error)
|
|
}
|
|
|
|
type httpInfrastructureMapper struct{}
|
|
|
|
var HTTPInfrastructure HTTPInfrastructurePersistenceInterface = httpInfrastructureMapper{}
|
|
|
|
func tenantHTTPInfrastructureAdapter(tenantID types.TenantID) (adapter.TenantHTTPInfrastructureAdapter, error) {
|
|
business, err := tenantBusinessAdapter(tenantID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
infrastructure, ok := business.(adapter.TenantHTTPInfrastructureAdapter)
|
|
if !ok {
|
|
return nil, types.ErrUnsupported
|
|
}
|
|
return infrastructure, nil
|
|
}
|
|
|
|
func (httpInfrastructureMapper) BeginIdempotency(tenantID types.TenantID, record *types.HTTPIdempotencyRecord) (*types.HTTPIdempotencyRecord, bool, error) {
|
|
infrastructure, err := tenantHTTPInfrastructureAdapter(tenantID)
|
|
if err != nil {
|
|
return nil, false, err
|
|
}
|
|
return infrastructure.TenantHTTPIdempotencyBegin(tenantID, record)
|
|
}
|
|
|
|
func (httpInfrastructureMapper) CompleteIdempotency(tenantID types.TenantID, record *types.HTTPIdempotencyRecord) error {
|
|
infrastructure, err := tenantHTTPInfrastructureAdapter(tenantID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return infrastructure.TenantHTTPIdempotencyComplete(tenantID, record)
|
|
}
|
|
|
|
func (httpInfrastructureMapper) AbortIdempotency(tenantID types.TenantID, record *types.HTTPIdempotencyRecord) error {
|
|
infrastructure, err := tenantHTTPInfrastructureAdapter(tenantID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return infrastructure.TenantHTTPIdempotencyAbort(tenantID, record)
|
|
}
|
|
|
|
func (httpInfrastructureMapper) BeginKafkaConsume(tenantID types.TenantID, record *types.KafkaConsumeRecord, deadline time.Time) (*types.KafkaConsumeRecord, bool, error) {
|
|
infrastructure, err := tenantHTTPInfrastructureAdapter(tenantID)
|
|
if err != nil {
|
|
return nil, false, err
|
|
}
|
|
return infrastructure.TenantKafkaConsumeBegin(tenantID, record, deadline)
|
|
}
|
|
|
|
func (httpInfrastructureMapper) MarkKafkaConsumeSucceeded(tenantID types.TenantID, eventID, consumerGroup string, now time.Time) error {
|
|
infrastructure, err := tenantHTTPInfrastructureAdapter(tenantID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return infrastructure.TenantKafkaConsumeSucceed(tenantID, eventID, consumerGroup, now)
|
|
}
|
|
|
|
func (httpInfrastructureMapper) MarkKafkaConsumeFailed(tenantID types.TenantID, eventID, consumerGroup, reason, message string, now time.Time) (*types.KafkaConsumeRecord, error) {
|
|
infrastructure, err := tenantHTTPInfrastructureAdapter(tenantID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return infrastructure.TenantKafkaConsumeFail(tenantID, eventID, consumerGroup, reason, message, now)
|
|
}
|
|
|
|
func (httpInfrastructureMapper) MarkKafkaConsumeDead(tenantID types.TenantID, dead *types.KafkaDeadLetter, now time.Time) error {
|
|
infrastructure, err := tenantHTTPInfrastructureAdapter(tenantID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return infrastructure.TenantKafkaConsumeDead(tenantID, dead, now)
|
|
}
|
|
|
|
func (httpInfrastructureMapper) ListKafkaDeadLetters(tenantID types.TenantID, opts types.KafkaDeadLetterListOptions) (*types.KafkaDeadLetterPage, error) {
|
|
infrastructure, err := tenantHTTPInfrastructureAdapter(tenantID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return infrastructure.TenantKafkaDeadLetterList(tenantID, opts)
|
|
}
|
|
|
|
func (httpInfrastructureMapper) GetKafkaDeadLetter(tenantID types.TenantID, deadID string) (*types.KafkaDeadLetter, error) {
|
|
infrastructure, err := tenantHTTPInfrastructureAdapter(tenantID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return infrastructure.TenantKafkaDeadLetterGet(tenantID, deadID)
|
|
}
|
|
|
|
func (httpInfrastructureMapper) MarkKafkaDeadLetter(tenantID types.TenantID, deadID, status, resolvedBy, note string, now time.Time) error {
|
|
infrastructure, err := tenantHTTPInfrastructureAdapter(tenantID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return infrastructure.TenantKafkaDeadLetterMark(tenantID, deadID, status, resolvedBy, note, now)
|
|
}
|
|
|
|
func (httpInfrastructureMapper) MarkKafkaDeadLetterReplay(tenantID types.TenantID, deadID, resolvedBy, note string, now time.Time) (*types.KafkaDeadLetter, error) {
|
|
infrastructure, err := tenantHTTPInfrastructureAdapter(tenantID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return infrastructure.TenantKafkaDeadLetterMarkReplay(tenantID, deadID, resolvedBy, note, now)
|
|
}
|