Push claw-wall messages into channel-memory

cmd/claw-wall/channel_memory.go

@@ -0,0 +1,173 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strings"
+	"time"
+)
+
+const (
+	channelMemorySourceKind    = "discord"
+	channelMemorySourceService = "claw-wall"
+	channelMemorySourceSurface = "discord"
+)
+
+type channelMemoryClient struct {
+	ingestURL string
+	token     string
+	client    *http.Client
+}
+
+type channelMemoryIngestRequest struct {
+	ChannelID string                     `json:"channel_id"`
+	Message   channelMemoryIngestMessage `json:"message"`
+	Source    channelMemoryIngestSource  `json:"source,omitempty"`
+	Metadata  map[string]string          `json:"metadata,omitempty"`
+	Scope     string                     `json:"scope,omitempty"`
+}
+
+type channelMemoryIngestMessage struct {
+	ID          string `json:"id"`
+	AuthorID    string `json:"author_id,omitempty"`
+	AuthorName  string `json:"author_name,omitempty"`
+	CreatedAt   string `json:"created_at,omitempty"`
+	EditedAt    string `json:"edited_at,omitempty"`
+	Deleted     bool   `json:"deleted,omitempty"`
+	Content     string `json:"content,omitempty"`
+	ContentHash string `json:"content_hash,omitempty"`
+}
+
+type channelMemoryIngestSource struct {
+	Kind    string `json:"kind,omitempty"`
+	Service string `json:"service,omitempty"`
+	Surface string `json:"surface,omitempty"`
+	GuildID string `json:"guild_id,omitempty"`
+}
+
+func newChannelMemoryClient(rawURL, token string, timeout time.Duration) (*channelMemoryClient, error) {
+	rawURL = strings.TrimSpace(rawURL)
+	if rawURL == "" {
+		return nil, nil
+	}
+	parsed, err := url.Parse(rawURL)
+	if err != nil {
+		return nil, fmt.Errorf("parse channel-memory ingest URL: %w", err)
+	}
+	if parsed.Scheme != "http" && parsed.Scheme != "https" {
+		return nil, fmt.Errorf("channel-memory ingest URL must use http or https")
+	}
+	if strings.TrimSpace(parsed.Host) == "" {
+		return nil, fmt.Errorf("channel-memory ingest URL must include a host")
+	}
+	if timeout <= 0 {
+		timeout = 2 * time.Second
+	}
+	return &channelMemoryClient{
+		ingestURL: rawURL,
+		token:     strings.TrimSpace(token),
+		client:    &http.Client{Timeout: timeout},
+	}, nil
+}
+
+func (c *channelMemoryClient) enabled() bool {
+	return c != nil && strings.TrimSpace(c.ingestURL) != ""
+}
+
+func (c *channelMemoryClient) ingestMessages(ctx context.Context, messages []wallMessage) (int, error) {
+	if !c.enabled() || len(messages) == 0 {
+		return 0, nil
+	}
+	pushed := 0
+	for _, msg := range messages {
+		if strings.TrimSpace(msg.ChannelID) == "" || strings.TrimSpace(msg.ID) == "" {
+			continue
+		}
+		if err := c.ingestMessage(ctx, msg); err != nil {
+			return pushed, err
+		}
+		pushed++
+	}
+	return pushed, nil
+}
+
+func (c *channelMemoryClient) ingestMessage(ctx context.Context, msg wallMessage) error {
+	payload := channelMemoryPayloadForMessage(msg)
+	body, err := json.Marshal(payload)
+	if err != nil {
+		return err
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.ingestURL, bytes.NewReader(body))
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	if c.token != "" {
+		req.Header.Set("Authorization", "Bearer "+c.token)
+	}
+
+	resp, err := c.client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 1024))
+		return fmt.Errorf("channel-memory ingest returned %s: %s", resp.Status, strings.TrimSpace(string(respBody)))
+	}
+	io.Copy(io.Discard, resp.Body)
+	return nil
+}
+
+func channelMemoryPayloadForMessage(msg wallMessage) channelMemoryIngestRequest {
+	scope := "channel:" + strings.TrimSpace(msg.ChannelID)
+	return channelMemoryIngestRequest{
+		ChannelID: strings.TrimSpace(msg.ChannelID),
+		Message: channelMemoryIngestMessage{
+			ID:          strings.TrimSpace(msg.ID),
+			AuthorID:    strings.TrimSpace(msg.AuthorID),
+			AuthorName:  strings.TrimSpace(msg.Author),
+			CreatedAt:   formatRFC3339(msg.Timestamp),
+			EditedAt:    formatRFC3339(msg.EditedAt),
+			Deleted:     msg.Deleted,
+			Content:     msg.Content,
+			ContentHash: wallMessageContentHash(msg),
+		},
+		Source: channelMemoryIngestSource{
+			Kind:    channelMemorySourceKind,
+			Service: channelMemorySourceService,
+			Surface: channelMemorySourceSurface,
+		},
+		Metadata: map[string]string{
+			"source_handle":    stableSourceHandle(msg),
+			"visibility_scope": scope,
+		},
+		Scope: scope,
+	}
+}
+
+func wallMessageContentHash(msg wallMessage) string {
+	seed := strings.Join([]string{
+		channelMemorySourceKind,
+		strings.TrimSpace(msg.ChannelID),
+		strings.TrimSpace(msg.ID),
+		msg.Content,
+		fmt.Sprintf("deleted=%t", msg.Deleted),
+	}, "\x00")
+	sum := sha256.Sum256([]byte(seed))
+	return "sha256:" + hex.EncodeToString(sum[:])
+}
+
+func formatRFC3339(ts time.Time) string {
+	if ts.IsZero() {
+		return ""
+	}
+	return ts.UTC().Format(time.RFC3339)
+}

cmd/claw-wall/discord.go

@@ -27,23 +27,26 @@ type discordPoller struct {
 	fetchLimit        int
 	backfillRetention time.Duration
 	backfillMaxPages  int
+	channelMemory     *channelMemoryClient
 	latestByPair      map[string]string
 	baseURL           string
 	cooldowns         *rateLimitTracker
 	now               func() time.Time
 }
 
 type discordAPIMessage struct {
-	ID          string              `json:"id"`
-	Content     string              `json:"content"`
-	Timestamp   string              `json:"timestamp"`
-	Author      discordAPIAuthor    `json:"author"`
-	Member      *discordAPIMember   `json:"member,omitempty"`
-	Attachments []discordAttachment `json:"attachments,omitempty"`
-	Embeds      []discordEmbed      `json:"embeds,omitempty"`
+	ID              string              `json:"id"`
+	Content         string              `json:"content"`
+	Timestamp       string              `json:"timestamp"`
+	EditedTimestamp *string             `json:"edited_timestamp,omitempty"`
+	Author          discordAPIAuthor    `json:"author"`
+	Member          *discordAPIMember   `json:"member,omitempty"`
+	Attachments     []discordAttachment `json:"attachments,omitempty"`
+	Embeds          []discordEmbed      `json:"embeds,omitempty"`
 }
 
 type discordAPIAuthor struct {
+	ID         string `json:"id"`
 	Username   string `json:"username"`
 	GlobalName string `json:"global_name"`
 }
@@ -160,7 +163,8 @@ func (p *discordPoller) pollOnce(ctx context.Context, logWriter io.Writer) {
 			continue
 		}
 		if len(messages) > 0 {
-			p.store.mergeAt(target.ChannelID, messages, p.now())
+			retained := p.store.mergeAt(target.ChannelID, messages, p.now())
+			p.pushChannelMemory(ctx, target.ChannelID, retained, logWriter)
 			p.recoverGapIfNeeded(ctx, target, latestID, messages, logWriter)
 		}
 		if strings.TrimSpace(newestID) != "" {
@@ -251,7 +255,7 @@ func (p *discordPoller) backfillAll(ctx context.Context, logWriter io.Writer) {
 			continue
 		}
 		p.store.setBackfillStatus(target.ChannelID, backfillStatusInProgress)
-		status, newestID, err := p.backfillChannel(ctx, target, p.backfillRetention, p.backfillMaxPages)
+		status, newestID, err := p.backfillChannel(ctx, target, p.backfillRetention, p.backfillMaxPages, logWriter)
 		if err != nil {
 			var rateLimitErr *discordRateLimitError
 			if errors.As(err, &rateLimitErr) {
@@ -269,7 +273,7 @@ func (p *discordPoller) backfillAll(ctx context.Context, logWriter io.Writer) {
 	}
 }
 
-func (p *discordPoller) backfillChannel(ctx context.Context, target tokenPair, retention time.Duration, maxPages int) (string, string, error) {
+func (p *discordPoller) backfillChannel(ctx context.Context, target tokenPair, retention time.Duration, maxPages int, logWriter io.Writer) (string, string, error) {
 	if retention <= 0 || maxPages <= 0 {
 		return backfillStatusUnavailable, "", nil
 	}
@@ -294,7 +298,8 @@ func (p *discordPoller) backfillChannel(ctx context.Context, target tokenPair, r
 		if compareSnowflakes(pageNewestID, newestID) > 0 {
 			newestID = pageNewestID
 		}
-		p.store.mergeAt(target.ChannelID, messages, p.now())
+		retained := p.store.mergeAt(target.ChannelID, messages, p.now())
+		p.pushChannelMemory(ctx, target.ChannelID, retained, logWriter)
 
 		oldestID := messages[0].ID
 		beforeID = oldestID
@@ -313,7 +318,7 @@ func (p *discordPoller) recoverGapIfNeeded(ctx context.Context, target tokenPair
 		return
 	}
 	oldestReturnedID := newestBatch[0].ID
-	status, err := p.backfillGap(ctx, target, previousLatestID, oldestReturnedID, p.backfillMaxPages)
+	status, err := p.backfillGap(ctx, target, previousLatestID, oldestReturnedID, p.backfillMaxPages, logWriter)
 	if err != nil {
 		var rateLimitErr *discordRateLimitError
 		if errors.As(err, &rateLimitErr) {
@@ -329,7 +334,7 @@ func (p *discordPoller) recoverGapIfNeeded(ctx context.Context, target tokenPair
 	}
 }
 
-func (p *discordPoller) backfillGap(ctx context.Context, target tokenPair, afterID, beforeID string, maxPages int) (string, error) {
+func (p *discordPoller) backfillGap(ctx context.Context, target tokenPair, afterID, beforeID string, maxPages int, logWriter io.Writer) (string, error) {
 	for page := 0; page < maxPages; page++ {
 		messages, _, err := p.fetchMessagesBefore(ctx, target, beforeID)
 		if err != nil {
@@ -353,7 +358,8 @@ func (p *discordPoller) backfillGap(ctx context.Context, target tokenPair, after
 			filtered = append(filtered, msg)
 		}
 		if len(filtered) > 0 {
-			p.store.mergeAt(target.ChannelID, filtered, p.now())
+			retained := p.store.mergeAt(target.ChannelID, filtered, p.now())
+			p.pushChannelMemory(ctx, target.ChannelID, retained, logWriter)
 		}
 		if reachedBoundary || len(messages) < p.fetchLimit {
 			return backfillStatusComplete, nil
@@ -363,6 +369,19 @@ func (p *discordPoller) backfillGap(ctx context.Context, target tokenPair, after
 	return backfillStatusPartial, nil
 }
 
+func (p *discordPoller) pushChannelMemory(ctx context.Context, channelID string, messages []wallMessage, logWriter io.Writer) {
+	if p.channelMemory == nil || len(messages) == 0 {
+		return
+	}
+	pushed, err := p.channelMemory.ingestMessages(ctx, messages)
+	if err == nil {
+		return
+	}
+	if logWriter != nil {
+		fmt.Fprintf(logWriter, "claw-wall: channel-memory ingest failed for channel %s after %d/%d messages: %v\n", channelID, pushed, len(messages), err)
+	}
+}
+
 func oldestMessageTime(messages []wallMessage) time.Time {
 	var oldest time.Time
 	for _, msg := range messages {
@@ -386,13 +405,21 @@ func convertDiscordMessage(channelID string, msg discordAPIMessage) (wallMessage
 	if err != nil {
 		timestamp = time.Time{}
 	}
+	var editedAt time.Time
+	if msg.EditedTimestamp != nil && strings.TrimSpace(*msg.EditedTimestamp) != "" {
+		if parsed, err := time.Parse(time.RFC3339, strings.TrimSpace(*msg.EditedTimestamp)); err == nil {
+			editedAt = parsed
+		}
+	}
 
 	return wallMessage{
 		ID:        messageID,
 		ChannelID: channelID,
+		AuthorID:  strings.TrimSpace(msg.Author.ID),
 		Author:    discordAuthorName(msg),
 		Content:   renderDiscordMessageContent(msg),
 		Timestamp: timestamp,
+		EditedAt:  editedAt,
 	}, true
 }