tech/sendico
2
0
Fork 0
Code Issues 31 Pull Requests Actions Packages Projects Releases Wiki Activity

Callbacks service docs updated

Browse Source
This commit is contained in:
Stephan D
2026-03-02 16:27:33 +01:00
parent 17e08ff26f
commit 2be76aa519
77 changed files with 803 additions and 764 deletions
Download Patch File Download Diff File Expand all files Collapse all files

121
api/edge/callbacks/internal/storage/service.go
View File

@@ -6,6 +6,7 @@ import (
"strings"
"time"
"github.com/tech/sendico/edge/callbacks/internal/model"
"github.com/tech/sendico/pkg/db"
"github.com/tech/sendico/pkg/db/repository"
"github.com/tech/sendico/pkg/db/repository/builder"
@@ -13,6 +14,7 @@ import (
"github.com/tech/sendico/pkg/db/storable"
"github.com/tech/sendico/pkg/merrors"
"github.com/tech/sendico/pkg/mlogger"
pmodel "github.com/tech/sendico/pkg/model"
"github.com/tech/sendico/pkg/mservice"
mutil "github.com/tech/sendico/pkg/mutil/db"
"go.mongodb.org/mongo-driver/v2/bson"
@@ -39,10 +41,10 @@ type mongoRepository struct {
}
type inboxDoc struct {
storable.Base `bson:",inline"`
EventID string `bson:"event_id"`
ClientID string `bson:"client_id"`
EventType string `bson:"event_type"`
storable.Base `bson:",inline"`
pmodel.OrganizationBoundBase `bson:",inline"`
EventID string `bson:"event_id"`
EventType string `bson:"event_type"`
}
func (d *inboxDoc) Collection() string {
@@ -64,12 +66,12 @@ type deliveryPolicy struct {
}
type endpointDoc struct {
storable.Base `bson:",inline"`
deliveryPolicy `bson:"retry_policy"`
ClientID string `bson:"client_id"`
Status string `bson:"status"`
URL string `bson:"url"`
EventTypes []string `bson:"event_types"`
storable.Base `bson:",inline"`
pmodel.OrganizationBoundBase `bson:",inline"`
deliveryPolicy `bson:"retry_policy"`
Status string `bson:"status"`
URL string `bson:"url"`
EventTypes []string `bson:"event_types"`
}
func (d *endpointDoc) Collection() string {
@@ -79,18 +81,18 @@ func (d *endpointDoc) Collection() string {
type taskDoc struct {
storable.Base `bson:",inline"`
deliveryPolicy `bson:"retry_policy"`
EventID string `bson:"event_id"`
EndpointID bson.ObjectID `bson:"endpoint_id"`
EndpointURL string `bson:"endpoint_url"`
Payload []byte `bson:"payload"`
Status TaskStatus `bson:"status"`
Attempt int `bson:"attempt"`
LastError string `bson:"last_error,omitempty"`
LastHTTPCode int `bson:"last_http_code,omitempty"`
NextAttemptAt time.Time `bson:"next_attempt_at"`
LockedUntil *time.Time `bson:"locked_until,omitempty"`
WorkerID string `bson:"worker_id,omitempty"`
DeliveredAt *time.Time `bson:"delivered_at,omitempty"`
EventID string `bson:"event_id"`
EndpointRef bson.ObjectID `bson:"endpoint_ref"`
EndpointURL string `bson:"endpoint_url"`
Payload []byte `bson:"payload"`
Status model.TaskStatus `bson:"status"`
Attempt int `bson:"attempt"`
LastError string `bson:"last_error,omitempty"`
LastHTTPCode int `bson:"last_http_code,omitempty"`
NextAttemptAt time.Time `bson:"next_attempt_at"`
LockedUntil *time.Time `bson:"locked_until,omitempty"`
WorkerID string `bson:"worker_id,omitempty"`
DeliveredAt *time.Time `bson:"delivered_at,omitempty"`
}
func (d *taskDoc) Collection() string {
@@ -152,7 +154,7 @@ func (m *mongoRepository) ensureIndexes() error {
Unique: true,
Keys: []ri.Key{
{Field: "event_id", Sort: ri.Asc},
{Field: "endpoint_id", Sort: ri.Asc},
{Field: "endpoint_ref", Sort: ri.Asc},
},
},
{
@@ -172,7 +174,7 @@ func (m *mongoRepository) ensureIndexes() error {
if err := m.endpointsRepo.CreateIndex(&ri.Definition{
Name: "idx_client_event",
Keys: []ri.Key{
{Field: "client_id", Sort: ri.Asc},
{Field: "organization_ref", Sort: ri.Asc},
{Field: "status", Sort: ri.Asc},
{Field: "event_types", Sort: ri.Asc},
},
@@ -188,11 +190,11 @@ type inboxStore struct {
repo repository.Repository
}
func (r *inboxStore) TryInsert(ctx context.Context, eventID, clientID, eventType string, at time.Time) (bool, error) {
func (r *inboxStore) TryInsert(ctx context.Context, eventID, eventType string, organizationRef bson.ObjectID, at time.Time) (bool, error) {
doc := &inboxDoc{
EventID: strings.TrimSpace(eventID),
ClientID: strings.TrimSpace(clientID),
EventType: strings.TrimSpace(eventType),
OrganizationBoundBase: pmodel.OrganizationBoundBase{OrganizationRef: organizationRef},
EventID: strings.TrimSpace(eventID),
EventType: strings.TrimSpace(eventType),
}
filter := repository.Filter("event_id", doc.EventID)
@@ -212,21 +214,20 @@ type endpointStore struct {
repo repository.Repository
}
func (r *endpointStore) FindActiveByClientAndType(ctx context.Context, clientID, eventType string) ([]Endpoint, error) {
clientID = strings.TrimSpace(clientID)
func (r *endpointStore) FindActive(ctx context.Context, eventType string, organizationRef bson.ObjectID) ([]model.Endpoint, error) {
eventType = strings.TrimSpace(eventType)
if clientID == "" {
return nil, merrors.InvalidArgument("client_id is required", "client_id")
if organizationRef == bson.NilObjectID {
return nil, merrors.InvalidArgument("organization_ref is required", "organization_ref")
}
if eventType == "" {
return nil, merrors.InvalidArgument("event type is required", "event_type")
}
query := repository.Query().
Filter(repository.Field("client_id"), clientID).
Filter(repository.OrgField(), organizationRef).
In(repository.Field("status"), "active", "enabled")
out := make([]Endpoint, 0)
out := make([]model.Endpoint, 0)
err := r.repo.FindManyByFilter(ctx, query, func(cur *mongo.Cursor) error {
doc := &endpointDoc{}
if err := cur.Decode(doc); err != nil {
@@ -238,17 +239,17 @@ func (r *endpointStore) FindActiveByClientAndType(ctx context.Context, clientID,
if !supportsEventType(doc.EventTypes, eventType) {
return nil
}
out = append(out, Endpoint{
ID: doc.ID,
ClientID: doc.ClientID,
URL: strings.TrimSpace(doc.URL),
SigningMode: strings.TrimSpace(doc.SigningMode),
SecretRef: strings.TrimSpace(doc.SecretRef),
Headers: cloneHeaders(doc.Headers),
MaxAttempts: doc.MaxAttempts,
MinDelay: time.Duration(doc.MinDelayMS) * time.Millisecond,
MaxDelay: time.Duration(doc.MaxDelayMS) * time.Millisecond,
RequestTimeout: time.Duration(doc.RequestTimeoutMS) * time.Millisecond,
out = append(out, model.Endpoint{
Base: doc.Base,
OrganizationBoundBase: doc.OrganizationBoundBase,
URL: strings.TrimSpace(doc.URL),
SigningMode: strings.TrimSpace(doc.SigningMode),
SecretRef: strings.TrimSpace(doc.SecretRef),
Headers: cloneHeaders(doc.Headers),
MaxAttempts: doc.MaxAttempts,
MinDelay: time.Duration(doc.MinDelayMS) * time.Millisecond,
MaxDelay: time.Duration(doc.MaxDelayMS) * time.Millisecond,
RequestTimeout: time.Duration(doc.RequestTimeoutMS) * time.Millisecond,
})
return nil
})
@@ -281,7 +282,7 @@ type taskStore struct {
repo repository.Repository
}
func (r *taskStore) UpsertTasks(ctx context.Context, eventID string, endpoints []Endpoint, payload []byte, defaults TaskDefaults, at time.Time) error {
func (r *taskStore) UpsertTasks(ctx context.Context, eventID string, endpoints []model.Endpoint, payload []byte, defaults TaskDefaults, at time.Time) error {
eventID = strings.TrimSpace(eventID)
if eventID == "" {
return merrors.InvalidArgument("event id is required", "event_id")
@@ -292,7 +293,7 @@ func (r *taskStore) UpsertTasks(ctx context.Context, eventID string, endpoints [
now := at.UTC()
for _, endpoint := range endpoints {
if endpoint.ID == bson.NilObjectID {
if endpoint.GetID() == nil || *endpoint.GetID() == bson.NilObjectID {
continue
}
@@ -327,13 +328,13 @@ func (r *taskStore) UpsertTasks(ctx context.Context, eventID string, endpoints [
doc := &taskDoc{}
doc.EventID = eventID
doc.EndpointID = endpoint.ID
doc.EndpointRef = *endpoint.GetID()
doc.EndpointURL = strings.TrimSpace(endpoint.URL)
doc.SigningMode = strings.TrimSpace(endpoint.SigningMode)
doc.SecretRef = strings.TrimSpace(endpoint.SecretRef)
doc.Headers = cloneHeaders(endpoint.Headers)
doc.Payload = append([]byte(nil), payload...)
doc.Status = TaskStatusPending
doc.Status = model.TaskStatusPending
doc.Attempt = 0
doc.MaxAttempts = maxAttempts
doc.MinDelayMS = int(minDelay / time.Millisecond)
@@ -341,7 +342,7 @@ func (r *taskStore) UpsertTasks(ctx context.Context, eventID string, endpoints [
doc.RequestTimeoutMS = int(requestTimeout / time.Millisecond)
doc.NextAttemptAt = now
filter := repository.Filter("event_id", eventID).And(repository.Filter("endpoint_id", endpoint.ID))
filter := repository.Filter("event_id", eventID).And(repository.Filter("endpoint_ref", endpoint.ID))
if err := r.repo.Insert(ctx, doc, filter); err != nil {
if errors.Is(err, merrors.ErrDataConflict) {
continue
@@ -353,7 +354,7 @@ func (r *taskStore) UpsertTasks(ctx context.Context, eventID string, endpoints [
return nil
}
func (r *taskStore) LockNextTask(ctx context.Context, now time.Time, workerID string, lockTTL time.Duration) (*Task, error) {
func (r *taskStore) LockNextTask(ctx context.Context, now time.Time, workerID string, lockTTL time.Duration) (*model.Task, error) {
workerID = strings.TrimSpace(workerID)
if workerID == "" {
return nil, merrors.InvalidArgument("worker id is required", "worker_id")
@@ -368,7 +369,7 @@ func (r *taskStore) LockNextTask(ctx context.Context, now time.Time, workerID st
)
query := repository.Query().
In(repository.Field("status"), string(TaskStatusPending), string(TaskStatusRetry)).
In(repository.Field("status"), string(model.TaskStatusPending), string(model.TaskStatusRetry)).
Comparison(repository.Field("next_attempt_at"), builder.Lte, now).
And(lockFilter).
Sort(repository.Field("next_attempt_at"), true).
@@ -390,7 +391,7 @@ func (r *taskStore) LockNextTask(ctx context.Context, now time.Time, workerID st
Set(repository.Field("worker_id"), workerID)
conditional := repository.IDFilter(candidate.ID).And(
repository.Query().In(repository.Field("status"), string(TaskStatusPending), string(TaskStatusRetry)),
repository.Query().In(repository.Field("status"), string(model.TaskStatusPending), string(model.TaskStatusRetry)),
repository.Query().Comparison(repository.Field("next_attempt_at"), builder.Lte, now),
lockFilter,
)
@@ -427,7 +428,7 @@ func (r *taskStore) MarkDelivered(ctx context.Context, taskID bson.ObjectID, htt
}
patch := repository.Patch().
Set(repository.Field("status"), TaskStatusDelivered).
Set(repository.Field("status"), model.TaskStatusDelivered).
Set(repository.Field("last_http_code"), httpCode).
Set(repository.Field("delivered_at"), time.Now()).
Set(repository.Field("locked_until"), nil).
@@ -446,7 +447,7 @@ func (r *taskStore) MarkRetry(ctx context.Context, taskID bson.ObjectID, attempt
}
patch := repository.Patch().
Set(repository.Field("status"), TaskStatusRetry).
Set(repository.Field("status"), model.TaskStatusRetry).
Set(repository.Field("attempt"), attempt).
Set(repository.Field("next_attempt_at"), nextAttemptAt.UTC()).
Set(repository.Field("last_error"), strings.TrimSpace(lastError)).
@@ -466,7 +467,7 @@ func (r *taskStore) MarkFailed(ctx context.Context, taskID bson.ObjectID, attemp
}
patch := repository.Patch().
Set(repository.Field("status"), TaskStatusFailed).
Set(repository.Field("status"), model.TaskStatusFailed).
Set(repository.Field("attempt"), attempt).
Set(repository.Field("last_error"), strings.TrimSpace(lastError)).
Set(repository.Field("last_http_code"), httpCode).
@@ -479,14 +480,14 @@ func (r *taskStore) MarkFailed(ctx context.Context, taskID bson.ObjectID, attemp
return nil
}
func mapTaskDoc(doc *taskDoc) *Task {
func mapTaskDoc(doc *taskDoc) *model.Task {
if doc == nil {
return nil
}
return &Task{
ID: doc.ID,
return &model.Task{
Base: doc.Base,
EventID: doc.EventID,
EndpointID: doc.EndpointID,
EndpointRef: doc.EndpointRef,
EndpointURL: doc.EndpointURL,
SigningMode: doc.SigningMode,
SecretRef: doc.SecretRef,