Merge pull request #534 from rusq/i533

Transformer unaccounted for channels with no threads

Author: rusq

internal/chunk/backend/dbase/repository/dbmessage_test.go

@@ -495,6 +495,8 @@ var (
 	tmC        = slack.Message{Msg: slack.Msg{Timestamp: "125.000", ThreadTimestamp: testThreadID, Text: "C"}}
 	tmD        = slack.Message{Msg: slack.Msg{Timestamp: "126.000", ThreadTimestamp: testThreadID, Text: "D"}}
 	tmC_       = slack.Message{Msg: slack.Msg{Timestamp: "125.000", ThreadTimestamp: testThreadID, Text: "C'"}}
+
+	mNoThread = slack.Message{Msg: slack.Msg{Timestamp: "130.000", Text: "No thread"}}
 	// special additional message to test the reference counter
 	tmXExtra = slack.Message{Msg: slack.Msg{Timestamp: "127.000", ThreadTimestamp: "127.000", Text: "X"}}
 	// thread lead that has replies deleted
@@ -510,6 +512,8 @@ var (
 	// these go into chunk 1
 	dbtmXExtra = must(NewDBMessage(1, 0, testChannelID, &tmXExtra))
 	dbtmYExtra = must(NewDBMessage(1, 0, testChannelID, &tmYExtra))
+
+	dbmNoThread = must(NewDBMessage(1, 0, testChannelID, &mNoThread))
 )
 
 // mkThreadSetupFn creates a utility function that sets up the thread messages.
@@ -827,6 +831,56 @@ func Test_messageRepository_CountUnfinished(t *testing.T) {
 				}
 			},
 		},
+		{
+			name: "channel with no threads #533, unfinished",
+			fields: fields{
+				genericRepository: genericRepository[DBMessage]{DBMessage{}},
+			},
+			args: args{
+				ctx:       t.Context(),
+				conn:      testConn(t),
+				sessionID: 1,
+				channelID: "C123",
+			},
+			prepFn: func(t *testing.T, conn PrepareExtContext) {
+				prepChunkWithFinal(
+					testChunk{typeID: chunk.CMessages, channelID: "C123", final: false},
+				)(t, conn)
+				mr := NewMessageRepository()
+				if err := mr.Insert(t.Context(), conn,
+					dbmNoThread,
+				); err != nil {
+					t.Fatalf("insert: %v", err)
+				}
+			},
+			want:    0,
+			wantErr: true, // no rows in the result set.  Set will be empty until there's a "final" message chunk for the channel.
+		},
+		{
+			name: "channel with no threads #533, finished",
+			fields: fields{
+				genericRepository: genericRepository[DBMessage]{DBMessage{}},
+			},
+			args: args{
+				ctx:       t.Context(),
+				conn:      testConn(t),
+				sessionID: 1,
+				channelID: "C123",
+			},
+			prepFn: func(t *testing.T, conn PrepareExtContext) {
+				prepChunkWithFinal(
+					testChunk{typeID: chunk.CMessages, channelID: "C123", final: true},
+				)(t, conn)
+				mr := NewMessageRepository()
+				if err := mr.Insert(t.Context(), conn,
+					dbmNoThread,
+				); err != nil {
+					t.Fatalf("insert: %v", err)
+				}
+			},
+			want:    0,
+			wantErr: false,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {

internal/chunk/backend/dbase/repository/migrations/20250628031633_include_no_thread_chans.sql

@@ -0,0 +1,58 @@
+-- +goose Up
+-- +goose StatementBegin
+DROP VIEW IF EXISTS V_CHANNEL_THREADS;
+-- +goose StatementEnd
+-- +goose StatementBegin
+CREATE VIEW IF NOT EXISTS V_CHANNEL_THREADS AS
+WITH MSG AS (SELECT ID, CHANNEL_ID, IS_PARENT, CHUNK_ID
+             FROM MESSAGE
+             EXCEPT
+             SELECT ID, CHANNEL_ID, IS_PARENT, CHUNK_ID
+             FROM MESSAGE
+             WHERE LATEST_REPLY = '0000000000.000000' -- EMPTY THREADS
+               AND IS_PARENT = TRUE)
+SELECT C.SESSION_ID, M.CHANNEL_ID, SUM(M.IS_PARENT) THREADS
+FROM MSG M
+         JOIN CHUNK C ON C.ID = M.CHUNK_ID
+WHERE 1 = 1 
+  AND C.TYPE_ID = 0
+  AND EXISTS (
+    -- CHANNEL MUST BE FINISHED
+    SELECT 1
+    FROM CHUNK MC
+    WHERE MC.CHANNEL_ID = C.CHANNEL_ID
+      AND MC.TYPE_ID = C.TYPE_ID
+      AND MC.SESSION_ID = C.SESSION_ID
+      AND FINAL)
+GROUP BY M.CHANNEL_ID
+-- +goose StatementEnd
+
+-- +goose Down
+-- +goose StatementBegin
+DROP VIEW IF EXISTS V_CHANNEL_THREADS;
+-- +goose StatementEnd
+-- +goose StatementBegin
+CREATE VIEW IF NOT EXISTS V_CHANNEL_THREADS AS
+WITH MSG AS (SELECT ID, CHANNEL_ID, IS_PARENT, CHUNK_ID
+             FROM MESSAGE
+             WHERE IS_PARENT = TRUE
+             EXCEPT
+             SELECT ID, CHANNEL_ID, IS_PARENT, CHUNK_ID
+             FROM MESSAGE
+             WHERE LATEST_REPLY = '0000000000.000000' -- EMPTY THREADS
+               AND IS_PARENT = TRUE)
+SELECT C.SESSION_ID, M.CHANNEL_ID, SUM(M.IS_PARENT) THREADS
+FROM MSG M
+         JOIN CHUNK C ON C.ID = M.CHUNK_ID
+WHERE IS_PARENT = TRUE
+  AND C.TYPE_ID = 0
+  AND EXISTS (
+    -- CHANNEL MUST BE FINISHED
+    SELECT 1
+    FROM CHUNK MC
+    WHERE MC.CHANNEL_ID = C.CHANNEL_ID
+      AND MC.TYPE_ID = C.TYPE_ID
+      AND MC.SESSION_ID = C.SESSION_ID
+      AND FINAL)
+GROUP BY M.CHANNEL_ID;
+-- +goose StatementEnd