-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
adyusupov
committed
May 29, 2024
1 parent
dd7cff7
commit 995c552
Showing
9 changed files
with
304 additions
and
75 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -29,7 +29,7 @@ jobs: | |
| version: latest | ||
|
|
||
| - name: Test | ||
| run: go test ./... | ||
| run: go clean --testcache && go test ./... | ||
|
|
||
| - name: Clean workspace | ||
| uses: AutoModality/[email protected] | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,16 +1,25 @@ | ||
| package clog | ||
|
|
||
| import "log/slog" | ||
| import ( | ||
| "log/slog" | ||
| "reflect" | ||
| ) | ||
|
|
||
| type fieldKey string | ||
|
|
||
| type fields map[fieldKey]interface{} | ||
| type Fields map[fieldKey]interface{} | ||
|
|
||
| // convertToAttrs converts a map of custom fields to a slice of slog.Attr | ||
| func convertToAttrs(fields fields) []any { | ||
| // ConvertToAttrs converts a map of custom fields to a slice of slog.Attr | ||
| func ConvertToAttrs(fields Fields) []any { | ||
| var attrs []any | ||
| for k, v := range fields { | ||
| attrs = append(attrs, slog.Any(string(k), v)) | ||
| if v != nil && !isZeroValue(v) { | ||
| attrs = append(attrs, slog.Any(string(k), v)) | ||
| } | ||
| } | ||
| return attrs | ||
| } | ||
|
|
||
| func isZeroValue(v interface{}) bool { | ||
| return v == reflect.Zero(reflect.TypeOf(v)).Interface() | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,45 @@ | ||
| package queue | ||
|
|
||
| import "context" | ||
|
|
||
| // loadEventsFromOutbox loads events from the outbox table into the in-memory queue. | ||
| func (bus *eventBus) loadEventsFromOutbox(ctx context.Context) error { | ||
| if bus.outbox == nil { | ||
| return nil | ||
| } | ||
|
|
||
| events, err := bus.outbox.LoadPendingEvents(ctx) | ||
| if err != nil { | ||
| bus.log.ErrorCtx(ctx, err, "Failed to load events from outbox") | ||
| return err | ||
| } | ||
|
|
||
| for _, outboxEvent := range events { | ||
| bus.queue <- convertOutboxEventToEvent(outboxEvent) | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| // updateEventStatus updates the status and retry count of an event in the outbox table. | ||
| func (bus *eventBus) updateEventStatus(ctx context.Context, event *Event) { | ||
| outboxEvent := convertEventToOutboxEvent(event) | ||
|
|
||
| if err := bus.outbox.UpdateEventStatus(ctx, outboxEvent); err != nil { | ||
| bus.log.ErrorCtx(ctx, err, "Failed to update event status in outbox") | ||
| } | ||
| } | ||
|
|
||
| // markEventAsProcessed marks an event as processed in the outbox table. | ||
| func (bus *eventBus) markEventAsProcessed(ctx context.Context, eventID int) { | ||
| if err := bus.outbox.MarkEventAsProcessed(ctx, eventID); err != nil { | ||
| bus.log.ErrorCtx(ctx, err, "Failed to mark event as processed in outbox") | ||
| } | ||
| } | ||
|
|
||
| // markEventAsFailed marks an event as failed in the outbox table. | ||
| func (bus *eventBus) markEventAsFailed(ctx context.Context, eventID int) { | ||
| if err := bus.outbox.MarkEventAsFailed(ctx, eventID); err != nil { | ||
| bus.log.ErrorCtx(ctx, err, "Failed to mark event as failed in outbox") | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,96 @@ | ||
| package queue | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
|
|
||
| "github.com/gateway-fm/scriptorium/transactions" | ||
| ) | ||
|
|
||
| // OutboxEvent represents an event stored in the outbox table. | ||
| type OutboxEvent struct { | ||
| ID int `pg:",pk"` // Primary key | ||
| Data []byte `pg:"data"` // Data column | ||
| Topic string `pg:"topic"` // Topic column | ||
| Retry int `pg:"retry"` // Retry column | ||
| NextRetry uint `pg:"next_retry"` // NextRetry column as minutes | ||
| AckStatus AckStatus `pg:"ack_status"` // AckStatus column | ||
| CreatedAt int64 `pg:"created_at"` // CreatedAt column as Unix timestamp | ||
| UpdatedAt int64 `pg:"updated_at"` // UpdatedAt column as Unix timestamp | ||
| } | ||
|
|
||
| // OutboxRepository provides methods to interact with the outbox table. | ||
| type OutboxRepository struct { | ||
| transactionFactory transactions.TransactionFactory | ||
| } | ||
|
|
||
| // NewOutboxRepository creates a new OutboxRepository. | ||
| func NewOutboxRepository(transactionFactory transactions.TransactionFactory) *OutboxRepository { | ||
| return &OutboxRepository{transactionFactory: transactionFactory} | ||
| } | ||
|
|
||
| // InsertEvent inserts a new event into the outbox table or updates it if it already exists. | ||
| func (r *OutboxRepository) InsertEvent(ctx context.Context, event *OutboxEvent) error { | ||
| _, err := r.transactionFactory.Transaction(ctx). | ||
| Model(event). | ||
| OnConflict("(data, topic) DO UPDATE"). | ||
| Set("retry = EXCLUDED.retry, next_retry = EXCLUDED.next_retry, ack_status = EXCLUDED.ack_status, updated_at = EXCLUDED.updated_at"). | ||
| Returning("*"). | ||
| Insert() | ||
| if err != nil { | ||
| return fmt.Errorf("insert event into outbox: %w", err) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // LoadPendingEvents loads all pending events from the outbox table. | ||
| func (r *OutboxRepository) LoadPendingEvents(ctx context.Context) ([]*OutboxEvent, error) { | ||
| var events []*OutboxEvent | ||
| query := r.transactionFactory.Transaction(ctx). | ||
| Model(&events). | ||
| Where("ack_status = ?", NACK) | ||
|
|
||
| if err := query.Select(); err != nil { | ||
| return nil, fmt.Errorf("load pending events from outbox: %w", err) | ||
| } | ||
| return events, nil | ||
| } | ||
|
|
||
| // UpdateEventStatus updates the status and retry count of an event in the outbox table. | ||
| func (r *OutboxRepository) UpdateEventStatus(ctx context.Context, event *OutboxEvent) error { | ||
| _, err := r.transactionFactory.Transaction(ctx). | ||
| Model(event). | ||
| Column("retry", "next_retry", "ack_status"). | ||
| Where("id = ?", event.ID). | ||
| Update() | ||
| if err != nil { | ||
| return fmt.Errorf("update event status in outbox: %w", err) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // MarkEventAsProcessed marks an event as processed in the outbox table. | ||
| func (r *OutboxRepository) MarkEventAsProcessed(ctx context.Context, eventID int) error { | ||
| _, err := r.transactionFactory.Transaction(ctx). | ||
| Model(&OutboxEvent{}). | ||
| Set("ack_status = ?", ACK). | ||
| Where("id = ?", eventID). | ||
| Update() | ||
| if err != nil { | ||
| return fmt.Errorf("mark event as processed in outbox: %w", err) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // MarkEventAsFailed marks an event as failed in the outbox table. | ||
| func (r *OutboxRepository) MarkEventAsFailed(ctx context.Context, eventID int) error { | ||
| _, err := r.transactionFactory.Transaction(ctx). | ||
| Model(&OutboxEvent{}). | ||
| Set("ack_status = ?", NACK). | ||
| Where("id = ?", eventID). | ||
| Update() | ||
| if err != nil { | ||
| return fmt.Errorf("mark event as failed in outbox: %w", err) | ||
| } | ||
| return nil | ||
| } |
Oops, something went wrong.