fix: use db clock for delaying event consumption (#1727)

We have a small delay between when a topic event is published and when
we allow subscriptions to consume it to guarantee no new events will be
inserted in-between.

Previously this check was done by the controller, comparing the db row's
created_at with the controller's clock.
It seems there are cases where the controller thinks that enough time
has passed, but not according to the db clock.
This PR moves these checks to the db layer. This should make our
integration test more reliable as it also relies on comparing two
timestamps generated by the db clock.

Example flaky integration test:
https://github.com/TBD54566975/ftl/actions/runs/9432792904/job/25983007930#step:8:1

backend/controller/dal/pubsub.go

@@ -58,15 +58,15 @@ func (d *DAL) ProgressSubscriptions(ctx context.Context, eventConsumptionDelay t
 
 	successful := 0
 	for _, subscription := range subs {
-		nextCursor, err := tx.db.GetNextEventForSubscription(ctx, subscription.Topic, subscription.Cursor)
+		nextCursor, err := tx.db.GetNextEventForSubscription(ctx, eventConsumptionDelay, subscription.Topic, subscription.Cursor)
 		if err != nil {
 			return 0, fmt.Errorf("failed to get next cursor: %w", translatePGError(err))
 		}
 		nextCursorKey, ok := nextCursor.Event.Get()
 		if !ok {
 			return 0, fmt.Errorf("could not find event to progress subscription: %w", translatePGError(err))
 		}
-		if nextCreatedAt, ok := nextCursor.CreatedAt.Get(); ok && nextCreatedAt.Add(eventConsumptionDelay).After(time.Now()) {
+		if !nextCursor.Ready {
 			logger.Tracef("Skipping subscription %s because event is too new", subscription.Key)
 			continue
 		}

backend/controller/sql/queries.sql

@@ -717,7 +717,8 @@ WITH cursor AS (
 )
 SELECT events."key" as event,
         events.payload,
-        events.created_at
+        events.created_at,
+        NOW() - events.created_at >= sqlc.arg('consumption_delay')::interval AS ready
 FROM topics
 LEFT JOIN topic_events as events ON events.topic_id = topics.id
 WHERE topics.key = sqlc.arg('topic')::topic_key