Clean up future events for cancelled timers for sqlite

Author: cschleiden

backend/sqlite/events.go

@@ -127,3 +127,14 @@ func insertEvents(ctx context.Context, tx *sql.Tx, tableName string, instanceID
 	}
 	return nil
 }
+
+func removeFutureEvent(ctx context.Context, tx *sql.Tx, instanceID string, scheduleEventID int64) error {
+	_, err := tx.ExecContext(
+		ctx,
+		"DELETE FROM `pending_events` WHERE instance_id = ? AND schedule_event_id = ? AND visible_at IS NOT NULL",
+		instanceID,
+		scheduleEventID,
+	)
+
+	return err
+}

backend/sqlite/schema.sql

@@ -25,7 +25,7 @@ CREATE TABLE IF NOT EXISTS `pending_events` (
   PRIMARY KEY(`id`, `instance_id`)
 );
 
-CREATE INDEX IF NOT EXISTS `idx_pending_events_instance_id_visible_at` ON `pending_events` (`instance_id`, `visible_at`);
+CREATE INDEX IF NOT EXISTS `idx_pending_events_instance_id_visible_at_schedule_event_id` ON `pending_events` (`instance_id`, `visible_at`, `schedule_event_id`);
 
 CREATE TABLE IF NOT EXISTS `history` (
   `id` TEXT,

backend/sqlite/sqlite.go

@@ -365,6 +365,15 @@ func (sb *sqliteBackend) CompleteWorkflowTask(
 		}
 	}
 
+	for _, event := range executedEvents {
+		switch event.Type {
+		case history.EventType_TimerCanceled:
+			if err := removeFutureEvent(ctx, tx, instance.InstanceID, event.ScheduleEventID); err != nil {
+				return fmt.Errorf("removing future event: %w", err)
+			}
+		}
+	}
+
 	// Insert new workflow events
 	groupedEvents := make(map[*workflow.Instance][]history.Event)
 	for _, m := range workflowEvents {

backend/sqlite/sqlite_test.go

@@ -2,6 +2,7 @@ package sqlite
 
 import (
 	"context"
+	"fmt"
 	"testing"
 
 	"github.com/cschleiden/go-workflows/backend"
@@ -26,6 +27,51 @@ func Test_EndToEndSqliteBackend(t *testing.T) {
 var _ test.TestBackend = (*sqliteBackend)(nil)
 
 func (sb *sqliteBackend) GetFutureEvents(ctx context.Context) ([]history.Event, error) {
-	// TODO: TESTING: Implement
-	return nil, nil
+	tx, err := sb.db.BeginTx(ctx, nil)
+	if err != nil {
+		return nil, err
+	}
+	defer tx.Rollback()
+
+	// There is no index on `visible_at`, but this is okay for test only usage.
+	futureEvents, err := tx.QueryContext(
+		ctx,
+		"SELECT id, sequence_id, instance_id, event_type, timestamp, schedule_event_id, attributes, visible_at FROM `pending_events` WHERE visible_at IS NOT NULL",
+	)
+	if err != nil {
+		return nil, fmt.Errorf("getting history: %w", err)
+	}
+
+	f := make([]history.Event, 0)
+
+	for futureEvents.Next() {
+		var instanceID string
+		var attributes []byte
+
+		fe := history.Event{}
+
+		if err := futureEvents.Scan(
+			&fe.ID,
+			&fe.SequenceID,
+			&instanceID,
+			&fe.Type,
+			&fe.Timestamp,
+			&fe.ScheduleEventID,
+			&attributes,
+			&fe.VisibleAt,
+		); err != nil {
+			return nil, fmt.Errorf("scanning event: %w", err)
+		}
+
+		a, err := history.DeserializeAttributes(fe.Type, attributes)
+		if err != nil {
+			return nil, fmt.Errorf("deserializing attributes: %w", err)
+		}
+
+		fe.Attributes = a
+
+		f = append(f, fe)
+	}
+
+	return f, nil
 }