add logging

Author: jsbroks

apps/workspace-engine/main.go

@@ -166,19 +166,23 @@ func main() {
 		log.Info("Kafka consumer started")
 		if err := kafka.RunConsumer(ctx); err != nil {
 			log.Error("received error from kafka consumer", "error", err)
+			panic(err)
 		}
 	}()
 
 	go func() {
 		log.Info("Workspace engine server started", "address", addr)
 		if err := router.Run(addr); err != nil {
 			log.Fatal("Failed to serve", "error", err)
+			panic(err)
 		}
 	}()
 
 	sigChan := make(chan os.Signal, 1)
 	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
+	
 	<-sigChan
 
+	producer.Close()
 	log.Info("Shutting down workspace engine...")
 }

apps/workspace-engine/pkg/events/handler/handler.go

@@ -167,6 +167,7 @@ func (el *EventListener) ListenAndRoute(ctx context.Context, msg *kafka.Message,
 
 	ctx = changeset.WithChangeSet(ctx, changeSet)
 
+	log.Info("Processing event", "eventType", rawEvent.EventType, "workspaceID", rawEvent.WorkspaceID)
 	if err := handler(ctx, ws, rawEvent); err != nil {
 		span.RecordError(err)
 		span.SetStatus(codes.Error, "handler failed")

apps/workspace-engine/pkg/events/handler/resources/resources.go

@@ -2,7 +2,6 @@ package resources
 
 import (
 	"context"
-	"errors"
 	"workspace-engine/pkg/events/handler"
 	"workspace-engine/pkg/oapi"
 	"workspace-engine/pkg/workspace"
@@ -20,9 +19,7 @@ func HandleResourceCreated(
 		return err
 	}
 
-	if resource.WorkspaceId != ws.ID {
-		return errors.New("resource workspace id does not match workspace id")
-	}
+	resource.WorkspaceId = ws.ID
 
 	if _, err := ws.Resources().Upsert(ctx, resource); err != nil {
 		return err

apps/workspace-engine/pkg/workspace/loader.go

@@ -73,7 +73,13 @@ func Load(ctx context.Context, storage StorageClient, workspace *Workspace) erro
 
 	data, err := storage.Get(ctx, dbSnapshot.Path)
 	if err != nil {
-		return fmt.Errorf("failed to read workspace from disk: %w", err)
+		span.RecordError(err)
+		span.SetStatus(codes.Error, "Failed to read workspace from disk")
+		log.Warn("Failed to read workspace from disk, populating workspace with initial state", "workspaceID", workspace.ID, "error", err)
+		if err := PopulateWorkspaceWithInitialState(ctx, workspace); err != nil {
+			return fmt.Errorf("failed to populate workspace with initial state: %w", err)
+		}
+		return nil
 	}
 
 	span.AddEvent("loaded workspace from disk")

packages/events/src/kafka/client.ts

@@ -51,12 +51,14 @@ export const sendGoEvent = createSpanWrapper(
           value: JSON.stringify(message),
         })),
       });
+      log.info("Sent event", { messages });
     } catch (error) {
       const err = error instanceof Error ? error : new Error(String(error));
       span.setStatus({
         code: SpanStatusCode.ERROR,
         message: err.message,
       });
+      log.error("Failed to send event", { error });
       throw error;
     }
   },