Merge branch 'main' into dlm/structured-logging

beam-cloud · Nov 25, 2024 · 62e35a4 · 62e35a4
2 parents 9bfecbf + 949d3aa
commit 62e35a4
Show file tree

Hide file tree

Showing 5 changed files with 46 additions and 5 deletions.
diff --git a/pkg/abstractions/experimental/bot/bot.go b/pkg/abstractions/experimental/bot/bot.go
@@ -338,6 +338,7 @@ var (
 	botInputBuffer      string = "bot:%s:%s:input_buffer:%s"
 	botEventPair        string = "bot:%s:%s:event_pair:%s:%s"
 	botEventBuffer      string = "bot:%s:%s:event_buffer:%s"
+	botEventHistory     string = "bot:%s:%s:event_history:%s"
 	botSessionIndex     string = "bot:%s:%s:session_index"
 	botSessionState     string = "bot:%s:%s:session_state:%s"
 	botSessionKeepAlive string = "bot:%s:%s:session_keep_alive:%s"
@@ -366,6 +367,10 @@ func (k *keys) botEventBuffer(workspaceName, stubId, sessionId string) string {
 	return fmt.Sprintf(botEventBuffer, workspaceName, stubId, sessionId)
 }
 
+func (k *keys) botEventHistory(workspaceName, stubId, sessionId string) string {
+	return fmt.Sprintf(botEventHistory, workspaceName, stubId, sessionId)
+}
+
 func (k *keys) botSessionState(workspaceName, stubId, sessionId string) string {
 	return fmt.Sprintf(botSessionState, workspaceName, stubId, sessionId)
 }

diff --git a/pkg/abstractions/experimental/bot/http.go b/pkg/abstractions/experimental/bot/http.go
@@ -39,11 +39,14 @@ func registerBotRoutes(g *echo.Group, pbs *PetriBotService) *botGroup {
 	g.GET("/:deploymentName", auth.WithAuth(group.BotOpenSession))
 	g.GET("/:deploymentName/latest", auth.WithAuth(group.BotOpenSession))
 	g.GET("/:deploymentName/v:version", auth.WithAuth(group.BotOpenSession))
-	g.GET("/public/:stubId", auth.WithAssumedStubAuth(group.BotOpenSession, group.pbs.isPublic))
 	g.DELETE("/:stubId/:sessionId", auth.WithAuth(group.BotDeleteSession))
 	g.GET("/:stubId/:sessionId", auth.WithAuth(group.BotGetSession))
 	g.GET("/:stubId/sessions", auth.WithAuth(group.BotListSessions))
 
+	// Public endpoints
+	g.GET("/public/:stubId", auth.WithAssumedStubAuth(group.BotOpenSession, group.pbs.isPublic))
+	g.GET("/public/:stubId/:sessionId", auth.WithAssumedStubAuth(group.BotGetSession, group.pbs.isPublic))
+
 	return group
 }
 

diff --git a/pkg/abstractions/experimental/bot/instance.go b/pkg/abstractions/experimental/bot/instance.go
@@ -375,6 +375,7 @@ func (i *botInstance) monitorEvents() error {
 						string(MetadataTransitionName): event.Metadata[string(MetadataTransitionName)],
 						string(MetadataTaskId):         event.Metadata[string(MetadataTaskId)],
 					},
+					PairId: event.PairId,
 				}, event)
 			case BotEventTypeAcceptTransition, BotEventTypeRejectTransition:
 				taskId := event.Metadata[string(MetadataTaskId)]

diff --git a/pkg/abstractions/experimental/bot/state.go b/pkg/abstractions/experimental/bot/state.go
@@ -89,6 +89,7 @@ func (m *botStateManager) deleteSession(workspaceName, stubId, sessionId string)
 	stateKey := Keys.botSessionState(workspaceName, stubId, sessionId)
 	indexKey := Keys.botSessionIndex(workspaceName, stubId)
 	taskIndexKey := Keys.botTaskIndex(workspaceName, stubId, sessionId)
+	historyKey := Keys.botEventHistory(workspaceName, stubId, sessionId)
 
 	err := m.rdb.Del(ctx, stateKey).Err()
 	if err != nil {
@@ -112,6 +113,11 @@ func (m *botStateManager) deleteSession(workspaceName, stubId, sessionId string)
 		return err
 	}
 
+	err = m.rdb.Del(ctx, historyKey).Err()
+	if err != nil {
+		return err
+	}
+
 	// TODO: delete all existing markers
 	return m.rdb.SRem(ctx, indexKey, sessionId).Err()
 }
@@ -145,6 +151,24 @@ func (m *botStateManager) getSession(workspaceName, stubId, sessionId string) (*
 		return nil, err
 	}
 
+	historyKey := Keys.botEventHistory(workspaceName, stubId, sessionId)
+	history, err := m.rdb.LRange(context.TODO(), historyKey, 0, -1)
+	if err != nil {
+		return nil, err
+	}
+
+	events := []*BotEvent{}
+	for _, jsonData := range history {
+		event := &BotEvent{}
+		err = json.Unmarshal([]byte(jsonData), event)
+		if err != nil {
+			return nil, err
+		}
+
+		events = append(events, event)
+	}
+
+	session.EventHistory = events
 	return session, nil
 }
 
@@ -266,7 +290,14 @@ func (m *botStateManager) pushEvent(workspaceName, stubId, sessionId string, eve
 	}
 
 	messageKey := Keys.botEventBuffer(workspaceName, stubId, sessionId)
-	return m.rdb.RPush(context.TODO(), messageKey, jsonData).Err()
+	historyKey := Keys.botEventHistory(workspaceName, stubId, sessionId)
+
+	err = m.rdb.RPush(context.TODO(), messageKey, jsonData).Err()
+	if err != nil {
+		return err
+	}
+
+	return m.rdb.RPush(context.TODO(), historyKey, jsonData).Err()
 }
 
 const eventPairTtlS = 120 * time.Second

diff --git a/pkg/abstractions/experimental/bot/types.go b/pkg/abstractions/experimental/bot/types.go
@@ -17,9 +17,10 @@ const (
 var ErrBotSessionNotFound = fmt.Errorf("bot session not found")
 
 type BotSession struct {
-	Id        string                     `json:"id" redis:"id"`
-	Messages  []BotChatCompletionMessage `json:"messages" redis:"messages"`
-	CreatedAt int64                      `json:"created_at" redis:"created_at"`
+	Id           string                     `json:"id" redis:"id"`
+	Messages     []BotChatCompletionMessage `json:"messages" redis:"messages"`
+	CreatedAt    int64                      `json:"created_at" redis:"created_at"`
+	EventHistory []*BotEvent                `json:"event_history" redis:"event_history"`
 }
 
 type botContainer struct {