goclaw/internal/channelmemory/review.go

package channelmemory

import (
	"context"
	"fmt"
	"time"

	"github.com/google/uuid"
	"github.com/nextlevelbuilder/goclaw/internal/eventbus"
	"github.com/nextlevelbuilder/goclaw/internal/store"
)

func (s *Service) Approve(ctx context.Context, itemID uuid.UUID, approver string) (*store.ChannelMemoryExtractionItem, error) {
	item, err := s.Extractions.GetItem(ctx, itemID)
	if err != nil {
		return nil, err
	}
	if item.Status == store.ChannelMemoryItemWritten {
		return item, nil
	}
	if item.Status != store.ChannelMemoryItemPendingReview && item.Status != store.ChannelMemoryItemApproved {
		return nil, fmt.Errorf("item is not approvable")
	}
	sourceID := item.SourceID
	// Passive channel extraction is reusable channel context, not a personal
	// memory of the channel instance creator or current sender.
	memoryUserID := ""
	exists, err := s.Episodic.ExistsBySourceID(ctx, item.AgentID.String(), memoryUserID, sourceID)
	if err != nil {
		return nil, err
	}
	if !exists {
		retention := s.retentionDuration(ctx, item)
		keyTopics := memoryKeyTopics(item)
		ep := &store.EpisodicSummary{
			TenantID:   item.TenantID,
			AgentID:    item.AgentID,
			UserID:     memoryUserID,
			SessionKey: "channel:" + item.ChannelInstanceID.String(),
			Summary:    item.Summary,
			KeyTopics:  keyTopics,
			L0Abstract: item.Summary,
			SourceID:   sourceID,
			SourceType: "channel",
			ExpiresAt:  timePtr(time.Now().UTC().Add(retention)),
		}
		if err := s.Episodic.Create(ctx, ep); err != nil {
			return nil, err
		}
		item.EpisodicID = ep.ID.String()
		if s.EventBus != nil {
			s.EventBus.Publish(eventbus.DomainEvent{
				Type:     eventbus.EventEpisodicCreated,
				SourceID: ep.ID.String(),
				TenantID: item.TenantID.String(),
				AgentID:  item.AgentID.String(),
				UserID:   item.UserID,
				Payload: &eventbus.EpisodicCreatedPayload{
					EpisodicID:  ep.ID.String(),
					SessionKey:  ep.SessionKey,
					Summary:     item.Summary,
					KeyTopics:   keyTopics,
					KeyEntities: decodeStrings(item.Entities),
				},
			})
		}
	} else {
		ep, err := s.Episodic.GetBySourceID(ctx, item.AgentID.String(), memoryUserID, sourceID)
		if err != nil {
			return nil, err
		}
		if ep != nil {
			item.EpisodicID = ep.ID.String()
		}
	}
	now := time.Now().UTC()
	if err := s.Extractions.UpdateItem(ctx, item.ID, map[string]any{
		"status":      store.ChannelMemoryItemWritten,
		"approved_by": approver,
		"approved_at": now,
		"written_at":  now,
		"episodic_id": item.EpisodicID,
	}); err != nil {
		return nil, err
	}
	item.Status = store.ChannelMemoryItemWritten
	return item, nil
}

func (s *Service) retentionDuration(ctx context.Context, item *store.ChannelMemoryExtractionItem) time.Duration {
	cfg := DefaultConfig()
	if s.Channels != nil && item != nil {
		inst, err := s.Channels.Get(ctx, item.ChannelInstanceID)
		if err == nil && inst != nil {
			cfg = ParseConfig(inst.Config)
		}
	}
	return time.Duration(cfg.RetentionHours) * time.Hour
}

func (s *Service) Reject(ctx context.Context, itemID uuid.UUID, actor string) error {
	return s.updateItemTerminal(ctx, itemID, store.ChannelMemoryItemRejected, map[string]any{"rejected_by": actor, "rejected_at": time.Now().UTC()})
}

func (s *Service) Delete(ctx context.Context, itemID uuid.UUID) error {
	item, err := s.Extractions.GetItem(ctx, itemID)
	if err != nil {
		return err
	}
	if item.EpisodicID != "" && s.Episodic != nil {
		if err := s.Episodic.Delete(ctx, item.EpisodicID); err != nil {
			return err
		}
	}
	return s.updateItemTerminal(ctx, itemID, store.ChannelMemoryItemDeleted, map[string]any{"deleted_at": time.Now().UTC()})
}

func (s *Service) updateItemTerminal(ctx context.Context, itemID uuid.UUID, status string, updates map[string]any) error {
	item, err := s.Extractions.GetItem(ctx, itemID)
	if err != nil {
		return err
	}
	if item.Status == store.ChannelMemoryItemWritten && status != store.ChannelMemoryItemDeleted {
		return fmt.Errorf("written item cannot transition to %s", status)
	}
	updates["status"] = status
	return s.Extractions.UpdateItem(ctx, itemID, updates)
}