bzzz/pkg/hmmm_adapter/smoke_test.go

package hmmm_adapter

import (
    "context"
    "encoding/json"
    "fmt"
    "sync"
    "testing"
    "time"
)

// TestPerIssueTopicSmokeTest tests the per-issue topic functionality without full BZZZ integration
func TestPerIssueTopicSmokeTest(t *testing.T) {
    // Mock pubsub functions that track calls
    joinedTopics := make(map[string]int)
    publishedMessages := make(map[string][]byte)
    var mu sync.Mutex

    joiner := func(topic string) error {
        mu.Lock()
        defer mu.Unlock()
        joinedTopics[topic]++
        return nil
    }

    publisher := func(topic string, payload []byte) error {
        mu.Lock()
        defer mu.Unlock()
        publishedMessages[topic] = payload
        return nil
    }

    adapter := NewAdapter(joiner, publisher)

    // Test per-issue topic publishing
    issueID := int64(42)
    topic := fmt.Sprintf("bzzz/meta/issue/%d", issueID)
    
    testMessage := map[string]interface{}{
        "version":    1,
        "type":       "meta_msg",
        "issue_id":   issueID,
        "thread_id":  "test-thread-42",
        "msg_id":     "smoke-test-msg-1",
        "node_id":    "test-node-id",
        "hop_count":  0,
        "timestamp":  time.Now().UTC(),
        "message":    "Smoke test: HMMM per-issue room initialized.",
    }

    payload, err := json.Marshal(testMessage)
    if err != nil {
        t.Fatalf("Failed to marshal test message: %v", err)
    }

    // Publish the message
    err = adapter.Publish(context.Background(), topic, payload)
    if err != nil {
        t.Fatalf("Failed to publish message: %v", err)
    }

    // Verify join was called once
    mu.Lock()
    if joinedTopics[topic] != 1 {
        t.Errorf("Expected topic %s to be joined once, got %d times", topic, joinedTopics[topic])
    }

    // Verify message was published
    if _, exists := publishedMessages[topic]; !exists {
        t.Errorf("Expected message to be published to topic %s", topic)
    }
    mu.Unlock()

    // Verify metrics
    metrics := adapter.GetMetrics()
    if metrics.PublishCount != 1 {
        t.Errorf("Expected publish count 1, got %d", metrics.PublishCount)
    }
    if metrics.JoinCount != 1 {
        t.Errorf("Expected join count 1, got %d", metrics.JoinCount)
    }
    if metrics.ErrorCount != 0 {
        t.Errorf("Expected error count 0, got %d", metrics.ErrorCount)
    }

    // Test publishing another message to the same topic (should not join again)
    testMessage2 := map[string]interface{}{
        "version":    1,
        "type":       "meta_msg",
        "issue_id":   issueID,
        "thread_id":  "test-thread-42",
        "msg_id":     "smoke-test-msg-2",
        "node_id":    "test-node-id",
        "hop_count":  0,
        "timestamp":  time.Now().UTC(),
        "message":    "Second message in same issue room.",
    }

    payload2, err := json.Marshal(testMessage2)
    if err != nil {
        t.Fatalf("Failed to marshal second test message: %v", err)
    }

    err = adapter.Publish(context.Background(), topic, payload2)
    if err != nil {
        t.Fatalf("Failed to publish second message: %v", err)
    }

    // Verify join was still called only once (topic cached)
    mu.Lock()
    if joinedTopics[topic] != 1 {
        t.Errorf("Expected topic %s to still be joined only once (cached), got %d times", topic, joinedTopics[topic])
    }
    mu.Unlock()

    // Verify updated metrics
    metrics = adapter.GetMetrics()
    if metrics.PublishCount != 2 {
        t.Errorf("Expected publish count 2, got %d", metrics.PublishCount)
    }
    if metrics.JoinCount != 1 {
        t.Errorf("Expected join count to remain 1 (cached), got %d", metrics.JoinCount)
    }

    t.Logf("✅ Per-issue topic smoke test passed: topic=%s, publishes=%d, joins=%d", 
        topic, metrics.PublishCount, metrics.JoinCount)
}

// TestMultiplePerIssueTopics tests publishing to multiple different per-issue topics
func TestMultiplePerIssueTopics(t *testing.T) {
    joinedTopics := make(map[string]int)
    publishedMessages := make(map[string][]byte)
    var mu sync.Mutex

    joiner := func(topic string) error {
        mu.Lock()
        defer mu.Unlock()
        joinedTopics[topic]++
        return nil
    }

    publisher := func(topic string, payload []byte) error {
        mu.Lock()
        defer mu.Unlock()
        publishedMessages[topic] = payload
        return nil
    }

    adapter := NewAdapter(joiner, publisher)

    // Test multiple issues
    issueIDs := []int64{100, 200, 300}
    
    for _, issueID := range issueIDs {
        topic := fmt.Sprintf("bzzz/meta/issue/%d", issueID)
        
        testMessage := map[string]interface{}{
            "version":    1,
            "type":       "meta_msg",
            "issue_id":   issueID,
            "thread_id":  fmt.Sprintf("issue-%d", issueID),
            "msg_id":     fmt.Sprintf("msg-%d-1", issueID),
            "node_id":    "test-node-id",
            "hop_count":  0,
            "timestamp":  time.Now().UTC(),
            "message":    fmt.Sprintf("Message for issue %d", issueID),
        }

        payload, err := json.Marshal(testMessage)
        if err != nil {
            t.Fatalf("Failed to marshal message for issue %d: %v", issueID, err)
        }

        err = adapter.Publish(context.Background(), topic, payload)
        if err != nil {
            t.Fatalf("Failed to publish message for issue %d: %v", issueID, err)
        }
    }

    // Verify all topics were joined once
    mu.Lock()
    for _, issueID := range issueIDs {
        topic := fmt.Sprintf("bzzz/meta/issue/%d", issueID)
        if joinedTopics[topic] != 1 {
            t.Errorf("Expected topic %s to be joined once, got %d times", topic, joinedTopics[topic])
        }
        if _, exists := publishedMessages[topic]; !exists {
            t.Errorf("Expected message to be published to topic %s", topic)
        }
    }
    mu.Unlock()

    // Verify metrics
    metrics := adapter.GetMetrics()
    expectedJoinCount := int64(len(issueIDs))
    expectedPublishCount := int64(len(issueIDs))

    if metrics.PublishCount != expectedPublishCount {
        t.Errorf("Expected publish count %d, got %d", expectedPublishCount, metrics.PublishCount)
    }
    if metrics.JoinCount != expectedJoinCount {
        t.Errorf("Expected join count %d, got %d", expectedJoinCount, metrics.JoinCount)
    }
    if metrics.ErrorCount != 0 {
        t.Errorf("Expected error count 0, got %d", metrics.ErrorCount)
    }

    // Verify all topics are cached
    cachedTopics := adapter.GetJoinedTopics()
    if len(cachedTopics) != len(issueIDs) {
        t.Errorf("Expected %d cached topics, got %d", len(issueIDs), len(cachedTopics))
    }

    t.Logf("✅ Multiple per-issue topics test passed: issues=%v, publishes=%d, joins=%d", 
        issueIDs, metrics.PublishCount, metrics.JoinCount)
}

// TestHMMMMessageFormat tests that the adapter can handle HMMM-formatted messages
func TestHMMMMessageFormat(t *testing.T) {
    joinedTopics := make(map[string]bool)
    var publishedPayload []byte
    var mu sync.Mutex

    joiner := func(topic string) error {
        mu.Lock()
        defer mu.Unlock()
        joinedTopics[topic] = true
        return nil
    }

    publisher := func(topic string, payload []byte) error {
        mu.Lock()
        defer mu.Unlock()
        publishedPayload = make([]byte, len(payload))
        copy(publishedPayload, payload)
        return nil
    }

    adapter := NewAdapter(joiner, publisher)

    // Create HMMM-compliant message (following HMMM message schema)
    hmmmMessage := map[string]interface{}{
        "version":    1,
        "type":       "meta_msg",
        "issue_id":   42,
        "thread_id":  "issue-42",
        "msg_id":     "seed-" + fmt.Sprintf("%d", time.Now().UnixNano()),
        "parent_id":  nil,
        "node_id":    "test-node-12D3KooW",
        "author":     "test-author",
        "hop_count":  0,
        "timestamp":  time.Now().UTC(),
        "message":    "Seed: HMMM per-issue room initialized.",
    }

    payload, err := json.Marshal(hmmmMessage)
    if err != nil {
        t.Fatalf("Failed to marshal HMMM message: %v", err)
    }

    topic := "bzzz/meta/issue/42"
    err = adapter.Publish(context.Background(), topic, payload)
    if err != nil {
        t.Fatalf("Failed to publish HMMM message: %v", err)
    }

    // Verify the message was published correctly
    mu.Lock()
    if !joinedTopics[topic] {
        t.Errorf("Expected topic %s to be joined", topic)
    }

    if len(publishedPayload) == 0 {
        t.Fatalf("Expected payload to be published")
    }

    // Unmarshal and verify the published payload matches the original
    var publishedMessage map[string]interface{}
    err = json.Unmarshal(publishedPayload, &publishedMessage)
    mu.Unlock()

    if err != nil {
        t.Fatalf("Failed to unmarshal published payload: %v", err)
    }

    // Verify key fields
    if publishedMessage["version"].(float64) != 1 {
        t.Errorf("Expected version 1, got %v", publishedMessage["version"])
    }
    if publishedMessage["type"].(string) != "meta_msg" {
        t.Errorf("Expected type 'meta_msg', got %v", publishedMessage["type"])
    }
    if publishedMessage["issue_id"].(float64) != 42 {
        t.Errorf("Expected issue_id 42, got %v", publishedMessage["issue_id"])
    }
    if publishedMessage["message"].(string) != "Seed: HMMM per-issue room initialized." {
        t.Errorf("Expected specific message, got %v", publishedMessage["message"])
    }

    t.Logf("✅ HMMM message format test passed: successfully published and parsed HMMM-compliant message")
}