feat: process messages for new channels at boot

any new channel added to config will get backfilled if there
are any messages to decrypt

closes: #68

db/sqlc/mock/querier.go

@@ -982,6 +982,21 @@ func (mr *MockQuerierMockRecorder) ListTraceTags(ctx, arg any) *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListTraceTags", reflect.TypeOf((*MockQuerier)(nil).ListTraceTags), ctx, arg)
 }
 
+// ListUndecryptedGroupTextPackets mocks base method.
+func (m *MockQuerier) ListUndecryptedGroupTextPackets(ctx context.Context) ([]db.ListUndecryptedGroupTextPacketsRow, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "ListUndecryptedGroupTextPackets", ctx)
+	ret0, _ := ret[0].([]db.ListUndecryptedGroupTextPacketsRow)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// ListUndecryptedGroupTextPackets indicates an expected call of ListUndecryptedGroupTextPackets.
+func (mr *MockQuerierMockRecorder) ListUndecryptedGroupTextPackets(ctx any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListUndecryptedGroupTextPackets", reflect.TypeOf((*MockQuerier)(nil).ListUndecryptedGroupTextPackets), ctx)
+}
+
 // ReconfirmNeighbors mocks base method.
 func (m *MockQuerier) ReconfirmNeighbors(ctx context.Context) error {
 	m.ctrl.T.Helper()

db/sqlc/querier.go

@@ -156,6 +156,11 @@ type Querier interface {
 	// IATA membership comes from trace_iatas (joining observations here spilled the
 	// hash join). Per-tag details filled in only for the returned page.
 	ListTraceTags(ctx context.Context, arg ListTraceTagsParams) ([]ListTraceTagsRow, error)
+	// Returns GRP_TXT packets (payload_type=5) never successfully decrypted. Used at boot to
+	// retry decryption against the current keystore for packets whose channel key was only added
+	// to the config after they'd already been ingested -- see
+	// internal/ingest.BackfillChannelMessages.
+	ListUndecryptedGroupTextPackets(ctx context.Context) ([]ListUndecryptedGroupTextPacketsRow, error)
 	// Delete node_neighbors where the neighbor has departed from node_short_ids
 	// for that IATA, or where its prefix_4 is now ambiguous.
 	ReconfirmNeighbors(ctx context.Context) error

db/sqlc/queries.sql.go

@@ -3029,6 +3029,40 @@ func (q *Queries) ListTraceTags(ctx context.Context, arg ListTraceTagsParams) ([
 	return items, nil
 }
 
+const listUndecryptedGroupTextPackets = `-- name: ListUndecryptedGroupTextPackets :many
+SELECT packet_hash, raw_payload FROM packets
+WHERE payload_type = 5 AND decrypted IS NOT TRUE
+`
+
+type ListUndecryptedGroupTextPacketsRow struct {
+	PacketHash []byte `json:"packet_hash"`
+	RawPayload []byte `json:"raw_payload"`
+}
+
+// Returns GRP_TXT packets (payload_type=5) never successfully decrypted. Used at boot to
+// retry decryption against the current keystore for packets whose channel key was only added
+// to the config after they'd already been ingested -- see
+// internal/ingest.BackfillChannelMessages.
+func (q *Queries) ListUndecryptedGroupTextPackets(ctx context.Context) ([]ListUndecryptedGroupTextPacketsRow, error) {
+	rows, err := q.db.Query(ctx, listUndecryptedGroupTextPackets)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	items := []ListUndecryptedGroupTextPacketsRow{}
+	for rows.Next() {
+		var i ListUndecryptedGroupTextPacketsRow
+		if err := rows.Scan(&i.PacketHash, &i.RawPayload); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
 const reconfirmNeighbors = `-- name: ReconfirmNeighbors :exec
 DELETE FROM node_neighbors nn
 WHERE NOT EXISTS (