feat(admin): manually retry an event (#27)

admin/api/api.go

@@ -21,10 +21,10 @@ type API struct {
 	cfg        *config.Config
 	log        *zap.SugaredLogger
 	DB         *db.DB
-	dispatcher dispatcher.Dispatcher
+	dispatcher *dispatcher.Dispatcher
 }
 
-func NewAPI(cfg *config.Config, db *db.DB, dispatcher dispatcher.Dispatcher) *API {
+func NewAPI(cfg *config.Config, db *db.DB, dispatcher *dispatcher.Dispatcher) *API {
 	return &API{
 		cfg:        cfg,
 		log:        zap.S(),
@@ -47,11 +47,13 @@ func (api *API) json(code int, w http.ResponseWriter, data interface{}) {
 	w.Header().Set("Content-Type", ApplicationJsonType)
 	w.WriteHeader(code)
 
-	bytes, err := json.Marshal(data)
-	if err != nil {
-		panic(err)
+	if data != nil {
+		bytes, err := json.Marshal(data)
+		if err != nil {
+			panic(err)
+		}
+		_, _ = w.Write(bytes)
 	}
-	_, _ = w.Write(bytes)
 }
 
 func (api *API) bindQuery(r *http.Request, q *query.Query) {
@@ -124,6 +126,7 @@ func (api *API) Handler() http.Handler {
 		r.HandleFunc(prefix+"/events", api.PageEvent).Methods("GET")
 		r.HandleFunc(prefix+"/events", api.CreateEvent).Methods("POST")
 		r.HandleFunc(prefix+"/events/{id}", api.GetEvent).Methods("GET")
+		r.HandleFunc(prefix+"/events/{id}/retry", api.RetryEvent).Methods("POST")
 	}
 
 	for _, prefix := range []string{"", "/workspaces/{workspace}"} {

admin/api/events.go

@@ -53,3 +53,26 @@ func (api *API) CreateEvent(w http.ResponseWriter, r *http.Request) {
 
 	api.json(201, w, event)
 }
+
+func (api *API) RetryEvent(w http.ResponseWriter, r *http.Request) {
+	id := api.param(r, "id")
+	event, err := api.DB.EventsWS.Get(r.Context(), id)
+	api.assert(err)
+	if event == nil {
+		api.json(404, w, ErrorResponse{Message: MsgNotFound})
+		return
+	}
+
+	endpointId := r.URL.Query().Get("endpoint_id")
+	endpoint, err := api.DB.EndpointsWS.Get(r.Context(), endpointId)
+	api.assert(err)
+	if endpoint == nil {
+		api.json(400, w, ErrorResponse{Message: "endpoint not found"})
+		return
+	}
+
+	err = api.dispatcher.DispatchEndpoint(r.Context(), id, []*entities.Endpoint{endpoint})
+	api.assert(err)
+
+	api.json(200, w, nil)
+}

app/app.go

@@ -32,7 +32,7 @@ type Application struct {
 	log        *zap.SugaredLogger
 	db         *db.DB
 	queue      queue.TaskQueue
-	dispatcher dispatcher.Dispatcher
+	dispatcher *dispatcher.Dispatcher
 	cache      cache.Cache
 
 	admin   *admin.Admin

db/dao/attempt_dao.go

@@ -17,6 +17,7 @@ type AttemptResult struct {
 	AttemptedAt types.Time
 	Status      entities.AttemptStatus
 	ErrorCode   *entities.AttemptErrorCode
+	Exhausted   bool
 }
 
 func NewAttemptDao(db *sqlx.DB, workspace bool) AttemptDAO {
@@ -32,6 +33,7 @@ func (dao *attemptDao) UpdateDelivery(ctx context.Context, id string, result *At
 		"attempted_at": result.AttemptedAt,
 		"status":       result.Status,
 		"error_code":   result.ErrorCode,
+		"exhausted":    result.Exhausted,
 	})
 	return err
 }

db/dao/dao.go

@@ -202,7 +202,11 @@ func (dao *DAO[T]) Insert(ctx context.Context, entity *T) error {
 		case "created_at", "updated_at": // ignore
 		default:
 			columns = append(columns, column)
-			values = append(values, v.Interface())
+			value := v.Interface()
+			if column == "ws_id" && dao.workspace {
+				value = ucontext.GetWorkspaceID(ctx)
+			}
+			values = append(values, value)
 		}
 	})
 	statement, args := psql.Insert(dao.table).Columns(columns...).Values(values...).
@@ -236,7 +240,11 @@ func (dao *DAO[T]) BatchInsert(ctx context.Context, entities []*T) error {
 			case "created_at", "updated_at":
 				// ignore
 			default:
-				values = append(values, v.Interface())
+				value := v.Interface()
+				if column == "ws_id" && dao.workspace {
+					value = ucontext.GetWorkspaceID(ctx)
+				}
+				values = append(values, value)
 			}
 		})
 		builder = builder.Values(values...)

db/entities/attempt.go

@@ -7,13 +7,15 @@ import (
 )
 
 type Attempt struct {
-	ID            string        `json:"id" db:"id"`
-	EventId       string        `json:"event_id" db:"event_id"`
-	EndpointId    string        `json:"endpoint_id" db:"endpoint_id"`
-	Status        AttemptStatus `json:"status" db:"status"`
-	AttemptNumber int           `json:"attempt_number" db:"attempt_number"`
-	ScheduledAt   types.Time    `json:"scheduled_at" db:"scheduled_at"`
-	AttemptedAt   *types.Time   `json:"attempted_at" db:"attempted_at"`
+	ID            string             `json:"id" db:"id"`
+	EventId       string             `json:"event_id" db:"event_id"`
+	EndpointId    string             `json:"endpoint_id" db:"endpoint_id"`
+	Status        AttemptStatus      `json:"status" db:"status"`
+	AttemptNumber int                `json:"attempt_number" db:"attempt_number"`
+	ScheduledAt   types.Time         `json:"scheduled_at" db:"scheduled_at"`
+	AttemptedAt   *types.Time        `json:"attempted_at" db:"attempted_at"`
+	TriggerMode   AttemptTriggerMode `json:"trigger_mode" db:"trigger_mode"`
+	Exhausted     bool               `json:"exhausted" db:"exhausted"`
 
 	ErrorCode *AttemptErrorCode `json:"error_code" db:"error_code"`
 	Request   *AttemptRequest   `json:"request" db:"request"`
@@ -41,6 +43,14 @@ const (
 	AttemptErrorCodeEndpointNotFound AttemptErrorCode = "ENDPOINT_NOT_FOUND"
 )
 
+type AttemptTriggerMode = string
+
+const (
+	AttemptTriggerModeInitial   AttemptTriggerMode = "INITIAL"
+	AttemptTriggerModeManual    AttemptTriggerMode = "MANUAL"
+	AttemptTriggerModeAutomatic AttemptTriggerMode = "AUTOMATIC"
+)
+
 type AttemptRequest struct {
 	Method  string            `json:"method"`
 	URL     string            `json:"url"`

db/migrations/2_attempts.down.sql

@@ -0,0 +1,2 @@
+ALTER TABLE IF EXISTS ONLY "attempts" DROP COLUMN IF EXISTS "trigger_mode";
+ALTER TABLE IF EXISTS ONLY "attempts" DROP COLUMN IF EXISTS "exhausted";

db/migrations/2_attempts.up.sql

@@ -0,0 +1,5 @@
+ALTER TABLE IF EXISTS ONLY "attempts" ADD COLUMN IF NOT EXISTS "trigger_mode" VARCHAR(10) NOT NULL;
+ALTER TABLE IF EXISTS ONLY "attempts" ADD COLUMN IF NOT EXISTS "exhausted" BOOLEAN NOT NULL DEFAULT false;
+
+UPDATE "attempts" SET trigger_mode = 'INITIAL' where attempt_number = 1;
+UPDATE "attempts" SET trigger_mode = 'AUTOMATIC' where attempt_number != 1;

dispatcher/dispatcher.go

@@ -14,38 +14,86 @@ import (
 )
 
 // Dispatcher is Event Dispatcher
-type Dispatcher interface {
-	Dispatch(ctx context.Context, event *entities.Event) error
-}
-
-type DBDispatcher struct {
+type Dispatcher struct {
 	log   *zap.SugaredLogger
 	queue queue.TaskQueue
 	db    *db.DB
 }
 
-func NewDispatcher(log *zap.SugaredLogger, queue queue.TaskQueue, db *db.DB) Dispatcher {
-
-	dispatcher := &DBDispatcher{
+func NewDispatcher(log *zap.SugaredLogger, queue queue.TaskQueue, db *db.DB) *Dispatcher {
+	dispatcher := &Dispatcher{
 		log:   log,
 		queue: queue,
 		db:    db,
 	}
 	return dispatcher
 }
 
-func (d *DBDispatcher) Dispatch(ctx context.Context, event *entities.Event) error {
+func (d *Dispatcher) Dispatch(ctx context.Context, event *entities.Event) error {
 	endpoints, err := listSubscribedEndpoints(ctx, d.db, event.EventType)
 	if err != nil {
 		return err
 	}
 
+	return d.dispatch(ctx, event, endpoints)
+}
+
+func (d *Dispatcher) DispatchEndpoint(ctx context.Context, eventId string, endpoints []*entities.Endpoint) error {
+	attempts := make([]*entities.Attempt, 0, len(endpoints))
+	tasks := make([]*queue.TaskMessage, 0, len(endpoints))
+
+	now := time.Now()
+	for _, endpoint := range endpoints {
+		delay := endpoint.Retry.Config.Attempts[0]
+		attempt := &entities.Attempt{
+			ID:            utils.KSUID(),
+			EventId:       eventId,
+			EndpointId:    endpoint.ID,
+			Status:        entities.AttemptStatusInit,
+			AttemptNumber: 1,
+			ScheduledAt:   types.NewTime(now.Add(time.Second * time.Duration(delay))),
+			TriggerMode:   entities.AttemptTriggerModeManual,
+		}
+
+		task := &queue.TaskMessage{
+			ID: attempt.ID,
+			Data: &model.MessageData{
+				EventID:    eventId,
+				EndpointId: endpoint.ID,
+				Attempt:    1,
+			},
+		}
+		attempts = append(attempts, attempt)
+		tasks = append(tasks, task)
+	}
+
+	err := d.db.AttemptsWS.BatchInsert(ctx, attempts)
+	if err != nil {
+		return err
+	}
+
+	for i, task := range tasks {
+		err := d.queue.Add(task, attempts[i].ScheduledAt.Time)
+		if err != nil {
+			d.log.Warnf("failed to add task to queue: %v", err)
+			continue
+		}
+		err = d.db.AttemptsWS.UpdateStatus(ctx, task.ID, entities.AttemptStatusQueued)
+		if err != nil {
+			d.log.Warnf("failed to update attempt status: %v", err)
+		}
+	}
+
+	return nil
+}
+
+func (d *Dispatcher) dispatch(ctx context.Context, event *entities.Event, endpoints []*entities.Endpoint) error {
 	attempts := make([]*entities.Attempt, 0, len(endpoints))
 	tasks := make([]*queue.TaskMessage, 0, len(endpoints))
 
-	err = d.db.TX(ctx, func(ctx context.Context) error {
+	err := d.db.TX(ctx, func(ctx context.Context) error {
 		now := time.Now()
-		err := d.db.Events.Insert(ctx, event)
+		err := d.db.EventsWS.Insert(ctx, event)
 		if err != nil {
 			return err
 		}
@@ -59,8 +107,8 @@ func (d *DBDispatcher) Dispatch(ctx context.Context, event *entities.Event) erro
 				Status:        entities.AttemptStatusInit,
 				AttemptNumber: 1,
 				ScheduledAt:   types.NewTime(now.Add(time.Second * time.Duration(delay))),
+				TriggerMode:   entities.AttemptTriggerModeInitial,
 			}
-			attempt.WorkspaceId = endpoint.WorkspaceId
 
 			task := &queue.TaskMessage{
 				ID: attempt.ID,

openapi.yml

@@ -258,7 +258,6 @@ paths:
                         items:
                           $ref: "#/components/schemas/Attempt"
 
-
   /workspaces/{ws_id}/attempts/{id}:
     parameters:
       - $ref: '#/components/parameters/workspace_id'
@@ -324,7 +323,6 @@ paths:
               schema:
                 $ref: '#/components/schemas/Event'
 
-
   /workspaces/{ws_id}/events/{id}:
     parameters:
       - $ref: '#/components/parameters/workspace_id'
@@ -347,6 +345,30 @@ paths:
               schema:
                 $ref: '#/components/schemas/Event'
 
+  /workspaces/{ws_id}/events/{id}/retry:
+    parameters:
+      - $ref: '#/components/parameters/workspace_id'
+
+    post:
+      summary: Manually retry an event
+      tags:
+        - Event
+      parameters:
+        - in: path
+          name: id
+          required: true
+          schema:
+            type: string
+        - in: query
+          name: endpoint_id
+          required: true
+          schema:
+            type: string
+
+      responses:
+        '200':
+          description: OK
+
 
   /workspaces/{ws_id}/sources:
     parameters:

proxy/gateway.go

@@ -30,10 +30,10 @@ type Gateway struct {
 	router *router.Router // TODO: happens-before
 	db     *db.DB
 
-	dispatcher dispatcher.Dispatcher
+	dispatcher *dispatcher.Dispatcher
 }
 
-func NewGateway(cfg *config.ProxyConfig, db *db.DB, dispatcher dispatcher.Dispatcher) *Gateway {
+func NewGateway(cfg *config.ProxyConfig, db *db.DB, dispatcher *dispatcher.Dispatcher) *Gateway {
 	gw := &Gateway{
 		cfg:        cfg,
 		log:        zap.S(),
@@ -111,7 +111,7 @@ func (gw *Gateway) Handle(w http.ResponseWriter, r *http.Request) {
 		})
 		return
 	}
-	event.WorkspaceId = source.WorkspaceId
+
 	err := gw.dispatcher.Dispatch(r.Context(), &event)
 	if err != nil {
 		w.Header().Set("Content-Type", "application/json")

test/admin/attempts_test.go

@@ -175,6 +175,8 @@ var _ = Describe("/attempts", Ordered, func() {
 					AttemptNumber: 1,
 					ScheduledAt:   types.Time{Time: time.Now()},
 					AttemptedAt:   &types.Time{Time: time.Now()},
+					TriggerMode:   entities.AttemptTriggerModeInitial,
+					Exhausted:     false,
 				}
 				entitiesConfig.Attempts = []*entities.Attempt{entity}
 
@@ -192,6 +194,8 @@ var _ = Describe("/attempts", Ordered, func() {
 				assert.Equal(GinkgoT(), entity.EventId, result.EventId)
 				assert.Equal(GinkgoT(), entity.EndpointId, result.EndpointId)
 				assert.Equal(GinkgoT(), entities.AttemptStatusSuccess, result.Status)
+				assert.Equal(GinkgoT(), entities.AttemptTriggerModeInitial, result.TriggerMode)
+				assert.Equal(GinkgoT(), false, result.Exhausted)
 				assert.EqualValues(GinkgoT(), 1, result.AttemptNumber)
 			})