Dev Server rerun UI (#1272)

pkg/coreapi/gql.schema.graphql

@@ -200,6 +200,7 @@ type FunctionRun {
   history: [RunHistoryItem!]!
   historyItemOutput(id: ULID!): String
   eventID: ID!
+  cron: String
 }
 
 enum HistoryType {

pkg/coreapi/graph/models/converters.go

@@ -72,6 +72,7 @@ func MakeFunctionRun(f *cqrs.FunctionRun) *FunctionRun {
 		EventID:    f.EventID.String(),
 		BatchID:    f.BatchID,
 		Status:     &status,
+		Cron:       f.Cron,
 	}
 	if len(f.Output) > 0 {
 		str := string(f.Output)

pkg/coreapi/graph/resolvers/function_run.resolver.go

@@ -10,7 +10,6 @@ import (
 
 	"github.com/google/uuid"
 	"github.com/inngest/inngest/pkg/coreapi/graph/models"
-	"github.com/inngest/inngest/pkg/cqrs"
 	"github.com/inngest/inngest/pkg/enums"
 	"github.com/inngest/inngest/pkg/event"
 	"github.com/inngest/inngest/pkg/execution"
@@ -280,18 +279,17 @@ func (r *mutationResolver) Rerun(
 	}
 
 	evt, err := r.Data.GetEventByInternalID(ctx, run.EventID)
-	if run.Cron != nil && err == sql.ErrNoRows {
-		// Create a dummy event since we don't store cron events. We can delete
-		// this dummy when we start storing cron events
-		evt = &cqrs.Event{}
-	} else if err != nil {
+	if err != nil {
 		return zero, fmt.Errorf("failed to get run event: %w", err)
 	}
 
 	identifier, err := r.Executor.Schedule(ctx, execution.ScheduleRequest{
 		Function: *fn,
 		Events: []event.TrackedEvent{
-			event.NewOSSTrackedEvent(evt.Event()),
+			// We need NewOSSTrackedEventWithID to ensure that the tracked event
+			// has the same ID as the original event. Calling NewOSSTrackedEvent
+			// will result in the creation of a new ID
+			event.NewOSSTrackedEventWithID(evt.Event(), evt.InternalID()),
 		},
 		OriginalRunID: &run.RunID,
 	})

pkg/coreapi/graph/resolvers/stream.go

@@ -73,44 +73,20 @@ func (r *queryResolver) Stream(ctx context.Context, q models.StreamQuery) ([]*mo
 			CreatedAt: time.UnixMilli(i.EventTS),
 			Runs:      []*models.FunctionRun{},
 		}
-		if len(fnsByID[i.ID]) > 0 {
-			items[n].Runs = fnsByID[i.ID]
-		}
-	}
-
-	// Query all function runs received, and filter by crons.
-	fns, err = r.Data.GetFunctionRunsTimebound(ctx, tb, q.Limit)
-	if err != nil {
-		return nil, err
-	}
-	for _, i := range fns {
-		if i.Cron == nil && i.OriginalRunID == nil {
-			// These are children of events.
-			continue
-		}
 
-		var trigger string
-		if i.Cron != nil {
-			trigger = *i.Cron
-		} else if i.OriginalRunID != nil {
-			trigger = "Cron rerun"
+		runs := fnsByID[i.ID]
+		if len(runs) > 0 {
+			// If any of the runs is a cron, then the stream item is a cron
+			for _, run := range runs {
+				if run.Cron != nil {
+					items[n].Trigger = *run.Cron
+					items[n].Type = models.StreamTypeCron
+					break
+				}
+			}
+
+			items[n].Runs = runs
 		}
-
-		runs := []*models.FunctionRun{models.MakeFunctionRun(i)}
-		_, err := r.Data.GetFunctionByInternalUUID(ctx, uuid.UUID{}, uuid.MustParse(runs[0].FunctionID))
-		if err == sql.ErrNoRows {
-			// Skip run since its function doesn't exist. This can happen when
-			// deleting a function or changing its ID.
-			runs = []*models.FunctionRun{}
-		}
-
-		items = append(items, &models.StreamItem{
-			ID:        i.RunID.String(),
-			Trigger:   trigger,
-			Type:      models.StreamTypeCron,
-			CreatedAt: i.RunStartedAt,
-			Runs:      runs,
-		})
 	}
 
 	sort.Slice(items, func(i, j int) bool {

pkg/devserver/devserver.go

@@ -250,6 +250,7 @@ func start(ctx context.Context, opts StartOpts) error {
 		runner.WithTracker(t),
 		runner.WithRateLimiter(rl),
 		runner.WithBatchManager(batcher),
+		runner.WithPublisher(pb),
 	)
 
 	// The devserver embeds the event API.

pkg/event/event.go

@@ -168,6 +168,13 @@ func NewOSSTrackedEvent(e Event) TrackedEvent {
 	}
 }
 
+func NewOSSTrackedEventWithID(e Event, id ulid.ULID) TrackedEvent {
+	return ossTrackedEvent{
+		Id:    id,
+		Event: e,
+	}
+}
+
 func NewOSSTrackedEventFromString(data string) (*ossTrackedEvent, error) {
 	evt := &ossTrackedEvent{}
 	if err := json.Unmarshal([]byte(data), evt); err != nil {

pkg/execution/runner/runner.go

@@ -3,6 +3,7 @@ package runner
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"fmt"
 	"sync"
 	"time"
@@ -106,6 +107,12 @@ func WithTracker(t *Tracker) func(s *svc) {
 	}
 }
 
+func WithPublisher(p pubsub.Publisher) func(s *svc) {
+	return func(s *svc) {
+		s.publisher = p
+	}
+}
+
 func NewService(c config.Config, opts ...Opt) Runner {
 	svc := &svc{config: c}
 	for _, o := range opts {
@@ -119,7 +126,8 @@ type svc struct {
 	cqrs   cqrs.Manager
 	// pubsub allows us to subscribe to new events, and re-publish events
 	// if there are errors.
-	pubsub pubsub.PublishSubscriber
+	pubsub    pubsub.PublishSubscriber
+	publisher pubsub.Publisher
 	// executor handles execution of functions.
 	executor execution.Executor
 	// data provides the required loading capabilities to trigger functions
@@ -252,13 +260,33 @@ func (s *svc) InitializeCrons(ctx context.Context) error {
 					))
 				defer span.End()
 
-				err := s.initialize(ctx, fn, event.NewOSSTrackedEvent(event.Event{
+				trackedEvent := event.NewOSSTrackedEvent(event.Event{
 					Data: map[string]any{
 						"cron": cron,
 					},
 					ID:   time.Now().UTC().Format(time.RFC3339),
 					Name: event.FnCronName,
-				}))
+				})
+
+				byt, err := json.Marshal(trackedEvent)
+				if err == nil {
+					err := s.publisher.Publish(
+						ctx,
+						s.config.EventStream.Service.TopicName(),
+						pubsub.Message{
+							Name:      event.EventReceivedName,
+							Data:      string(byt),
+							Timestamp: time.Now(),
+						},
+					)
+					if err != nil {
+						logger.From(ctx).Error().Err(err).Msg("error publishing cron event")
+					}
+				} else {
+					logger.From(ctx).Error().Err(err).Msg("error marshaling cron event")
+				}
+
+				err = s.initialize(ctx, fn, trackedEvent)
 				if err != nil {
 					logger.From(ctx).Error().Err(err).Msg("error initializing scheduled function")
 				}

pkg/history_drivers/memory_writer/writer.go

@@ -88,6 +88,7 @@ func (w *writer) writeWorkflowStart(
 	run := w.store.Data[item.RunID]
 	run.Run.AccountID = item.AccountID
 	run.Run.BatchID = item.BatchID
+	run.Run.Cron = item.Cron
 	run.Run.EventID = item.EventID
 	run.Run.ID = item.RunID
 	run.Run.OriginalRunID = item.OriginalRunID