973 lines
33 KiB
Go
973 lines
33 KiB
Go
package sqlite
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"chat-switchboard/models"
|
|
"chat-switchboard/store"
|
|
)
|
|
|
|
type ChannelStore struct{}
|
|
|
|
func NewChannelStore() *ChannelStore { return &ChannelStore{} }
|
|
|
|
func (s *ChannelStore) Create(ctx context.Context, ch *models.Channel) error {
|
|
ch.ID = store.NewID()
|
|
now := time.Now().UTC()
|
|
ch.CreatedAt = now
|
|
ch.UpdatedAt = now
|
|
_, err := DB.ExecContext(ctx, `
|
|
INSERT INTO channels (id, user_id, title, description, type, model, system_prompt,
|
|
provider_config_id, is_archived, is_pinned, folder_id, team_id, settings, created_at, updated_at)
|
|
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
|
|
ch.ID, ch.UserID, ch.Title, ch.Description, ch.Type, ch.Model, ch.SystemPrompt,
|
|
models.NullString(ch.ProviderConfigID), ch.IsArchived, ch.IsPinned,
|
|
models.NullString(ch.FolderID), models.NullString(ch.TeamID), ToJSON(ch.Settings),
|
|
now.Format(timeFmt), now.Format(timeFmt),
|
|
)
|
|
return err
|
|
}
|
|
|
|
func (s *ChannelStore) GetByID(ctx context.Context, id string) (*models.Channel, error) {
|
|
var ch models.Channel
|
|
var providerConfigID, folderID, teamID sql.NullString
|
|
var desc sql.NullString
|
|
var settingsJSON []byte
|
|
err := DB.QueryRowContext(ctx, `
|
|
SELECT id, user_id, title, description, type, model, system_prompt,
|
|
provider_config_id, is_archived, is_pinned, folder_id, team_id, settings,
|
|
created_at, updated_at
|
|
FROM channels WHERE id = ?`, id).Scan(
|
|
&ch.ID, &ch.UserID, &ch.Title, &desc, &ch.Type, &ch.Model, &ch.SystemPrompt,
|
|
&providerConfigID, &ch.IsArchived, &ch.IsPinned, &folderID, &teamID, &settingsJSON,
|
|
st(&ch.CreatedAt), st(&ch.UpdatedAt),
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
ch.Description = NullableString(desc)
|
|
ch.ProviderConfigID = NullableStringPtr(providerConfigID)
|
|
ch.FolderID = NullableStringPtr(folderID)
|
|
ch.TeamID = NullableStringPtr(teamID)
|
|
json.Unmarshal(settingsJSON, &ch.Settings)
|
|
return &ch, nil
|
|
}
|
|
|
|
func (s *ChannelStore) Update(ctx context.Context, id string, fields map[string]interface{}) error {
|
|
b := NewUpdate("channels")
|
|
for k, v := range fields {
|
|
if k == "settings" || k == "tags" {
|
|
b.SetJSON(k, v)
|
|
} else {
|
|
b.Set(k, v)
|
|
}
|
|
}
|
|
if !b.HasSets() {
|
|
return nil
|
|
}
|
|
b.Where("id", id)
|
|
_, err := b.Exec(DB)
|
|
return err
|
|
}
|
|
|
|
func (s *ChannelStore) Delete(ctx context.Context, id string) error {
|
|
_, err := DB.ExecContext(ctx, "DELETE FROM channels WHERE id = ?", id)
|
|
return err
|
|
}
|
|
|
|
func (s *ChannelStore) ListForUser(ctx context.Context, userID string, opts store.ListOptions) ([]models.Channel, int, error) {
|
|
// Count
|
|
var total int
|
|
DB.QueryRowContext(ctx, "SELECT COUNT(*) FROM channels WHERE user_id = ?", userID).Scan(&total)
|
|
|
|
b := NewSelect(
|
|
"id, user_id, title, description, type, model, system_prompt, provider_config_id, is_archived, is_pinned, folder_id, team_id, settings, created_at, updated_at",
|
|
"channels",
|
|
).Where("user_id = ?", userID)
|
|
|
|
if opts.Sort == "" {
|
|
b.OrderBy("updated_at", "DESC")
|
|
}
|
|
b.Paginate(opts)
|
|
|
|
q, args := b.Build()
|
|
rows, err := DB.QueryContext(ctx, q, args...)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var result []models.Channel
|
|
for rows.Next() {
|
|
var ch models.Channel
|
|
var providerConfigID, folderID, teamID, desc sql.NullString
|
|
var settingsJSON []byte
|
|
err := rows.Scan(&ch.ID, &ch.UserID, &ch.Title, &desc, &ch.Type, &ch.Model,
|
|
&ch.SystemPrompt, &providerConfigID, &ch.IsArchived, &ch.IsPinned,
|
|
&folderID, &teamID, &settingsJSON, st(&ch.CreatedAt), st(&ch.UpdatedAt))
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
ch.Description = NullableString(desc)
|
|
ch.ProviderConfigID = NullableStringPtr(providerConfigID)
|
|
ch.FolderID = NullableStringPtr(folderID)
|
|
ch.TeamID = NullableStringPtr(teamID)
|
|
json.Unmarshal(settingsJSON, &ch.Settings)
|
|
result = append(result, ch)
|
|
}
|
|
return result, total, rows.Err()
|
|
}
|
|
|
|
func (s *ChannelStore) Search(ctx context.Context, userID, query string, opts store.ListOptions) ([]models.Channel, int, error) {
|
|
// Simple title search for now — will add full-text when search feature lands
|
|
b := NewSelect(
|
|
"id, user_id, title, description, type, model, system_prompt, provider_config_id, is_archived, is_pinned, folder_id, team_id, settings, created_at, updated_at",
|
|
"channels",
|
|
).Where("user_id = ?", userID).Where("title LIKE ?", "%"+query+"%")
|
|
b.OrderBy("updated_at", "DESC")
|
|
b.Paginate(opts)
|
|
|
|
var total int
|
|
DB.QueryRowContext(ctx,
|
|
"SELECT COUNT(*) FROM channels WHERE user_id = ? AND title LIKE ?",
|
|
userID, "%"+query+"%").Scan(&total)
|
|
|
|
q, args := b.Build()
|
|
rows, err := DB.QueryContext(ctx, q, args...)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var result []models.Channel
|
|
for rows.Next() {
|
|
var ch models.Channel
|
|
var providerConfigID, folderID, teamID, desc sql.NullString
|
|
var settingsJSON []byte
|
|
err := rows.Scan(&ch.ID, &ch.UserID, &ch.Title, &desc, &ch.Type, &ch.Model,
|
|
&ch.SystemPrompt, &providerConfigID, &ch.IsArchived, &ch.IsPinned,
|
|
&folderID, &teamID, &settingsJSON, st(&ch.CreatedAt), st(&ch.UpdatedAt))
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
ch.Description = NullableString(desc)
|
|
ch.ProviderConfigID = NullableStringPtr(providerConfigID)
|
|
ch.FolderID = NullableStringPtr(folderID)
|
|
ch.TeamID = NullableStringPtr(teamID)
|
|
json.Unmarshal(settingsJSON, &ch.Settings)
|
|
result = append(result, ch)
|
|
}
|
|
return result, total, rows.Err()
|
|
}
|
|
|
|
func (s *ChannelStore) GetCursor(ctx context.Context, channelID, userID string) (*models.ChannelCursor, error) {
|
|
var c models.ChannelCursor
|
|
var leafID sql.NullString
|
|
err := DB.QueryRowContext(ctx, `
|
|
SELECT id, channel_id, user_id, active_leaf_id, updated_at
|
|
FROM channel_cursors WHERE channel_id = ? AND user_id = ?`,
|
|
channelID, userID).Scan(&c.ID, &c.ChannelID, &c.UserID, &leafID, st(&c.UpdatedAt))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
c.ActiveLeafID = NullableStringPtr(leafID)
|
|
return &c, nil
|
|
}
|
|
|
|
func (s *ChannelStore) SetCursor(ctx context.Context, channelID, userID, leafID string) error {
|
|
_, err := DB.ExecContext(ctx, `
|
|
INSERT INTO channel_cursors (id, channel_id, user_id, active_leaf_id)
|
|
VALUES (?, ?, ?, ?)
|
|
ON CONFLICT (channel_id, user_id) DO UPDATE SET active_leaf_id = excluded.active_leaf_id, updated_at = datetime('now')`,
|
|
store.NewID(), channelID, userID, leafID)
|
|
return err
|
|
}
|
|
|
|
func (s *ChannelStore) SetModel(ctx context.Context, cm *models.ChannelModel) error {
|
|
// For persona entries, use the persona-aware path
|
|
if cm.PersonaID != nil && *cm.PersonaID != "" {
|
|
return s.SetPersonaModel(ctx, cm)
|
|
}
|
|
// Nullable provider_config_id: empty string → SQL NULL (UUID FK)
|
|
var provCfgID interface{} = cm.ProviderConfigID
|
|
if cm.ProviderConfigID == "" {
|
|
provCfgID = nil
|
|
}
|
|
// Raw model upsert (no persona) — matches idx_channel_models_raw partial index
|
|
_, err := DB.ExecContext(ctx, `
|
|
INSERT INTO channel_models (id, channel_id, model_id, provider_config_id, display_name, system_prompt, settings, is_default)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT (channel_id, model_id, provider_config_id) WHERE persona_id IS NULL DO UPDATE SET
|
|
display_name = excluded.display_name, system_prompt = excluded.system_prompt, settings = excluded.settings, is_default = excluded.is_default`,
|
|
store.NewID(), cm.ChannelID, cm.ModelID, provCfgID, cm.DisplayName, cm.SystemPrompt, "{}", cm.IsDefault)
|
|
return err
|
|
}
|
|
|
|
// SetPersonaModel inserts or updates a persona's channel model roster entry.
|
|
func (s *ChannelStore) SetPersonaModel(ctx context.Context, cm *models.ChannelModel) error {
|
|
_, err := DB.ExecContext(ctx, `
|
|
INSERT INTO channel_models (id, channel_id, model_id, provider_config_id, persona_id, display_name, system_prompt, settings, is_default)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT (channel_id, persona_id) WHERE persona_id IS NOT NULL DO UPDATE SET
|
|
model_id = excluded.model_id, provider_config_id = excluded.provider_config_id, display_name = excluded.display_name, system_prompt = excluded.system_prompt, settings = excluded.settings, is_default = excluded.is_default`,
|
|
store.NewID(), cm.ChannelID, cm.ModelID, cm.ProviderConfigID, cm.PersonaID, cm.DisplayName, cm.SystemPrompt, "{}", cm.IsDefault)
|
|
return err
|
|
}
|
|
|
|
func (s *ChannelStore) GetModels(ctx context.Context, channelID string) ([]models.ChannelModel, error) {
|
|
rows, err := DB.QueryContext(ctx, `
|
|
SELECT cm.id, cm.channel_id, cm.model_id, COALESCE(cm.provider_config_id, ''),
|
|
COALESCE(cm.persona_id, ''), COALESCE(p.handle, ''), COALESCE(cm.display_name, ''),
|
|
COALESCE(cm.system_prompt, ''), cm.is_default
|
|
FROM channel_models cm
|
|
LEFT JOIN personas p ON p.id = cm.persona_id
|
|
WHERE cm.channel_id = ?`, channelID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var result []models.ChannelModel
|
|
for rows.Next() {
|
|
var cm models.ChannelModel
|
|
var personaID string
|
|
if err := rows.Scan(&cm.ID, &cm.ChannelID, &cm.ModelID, &cm.ProviderConfigID,
|
|
&personaID, &cm.Handle, &cm.DisplayName, &cm.SystemPrompt, &cm.IsDefault); err != nil {
|
|
return nil, err
|
|
}
|
|
if personaID != "" {
|
|
cm.PersonaID = &personaID
|
|
}
|
|
result = append(result, cm)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
func (s *ChannelStore) GetModelByID(ctx context.Context, id string) (*models.ChannelModel, error) {
|
|
var cm models.ChannelModel
|
|
var personaID string
|
|
err := DB.QueryRowContext(ctx, `
|
|
SELECT cm.id, cm.channel_id, cm.model_id, COALESCE(cm.provider_config_id, ''),
|
|
COALESCE(cm.persona_id, ''), COALESCE(p.handle, ''), COALESCE(cm.display_name, ''),
|
|
COALESCE(cm.system_prompt, ''), cm.is_default
|
|
FROM channel_models cm
|
|
LEFT JOIN personas p ON p.id = cm.persona_id
|
|
WHERE cm.id = ?`, id).Scan(
|
|
&cm.ID, &cm.ChannelID, &cm.ModelID, &cm.ProviderConfigID,
|
|
&personaID, &cm.Handle, &cm.DisplayName, &cm.SystemPrompt, &cm.IsDefault)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if personaID != "" {
|
|
cm.PersonaID = &personaID
|
|
}
|
|
return &cm, nil
|
|
}
|
|
|
|
func (s *ChannelStore) UpdateModel(ctx context.Context, id string, fields map[string]interface{}) error {
|
|
if len(fields) == 0 {
|
|
return nil
|
|
}
|
|
sets := make([]string, 0, len(fields))
|
|
args := make([]interface{}, 0, len(fields)+1)
|
|
for col, val := range fields {
|
|
sets = append(sets, col+" = ?")
|
|
args = append(args, val)
|
|
}
|
|
args = append(args, id)
|
|
query := "UPDATE channel_models SET " + strings.Join(sets, ", ") + " WHERE id = ?"
|
|
_, err := DB.ExecContext(ctx, query, args...)
|
|
return err
|
|
}
|
|
|
|
func (s *ChannelStore) DeleteModel(ctx context.Context, id string) error {
|
|
res, err := DB.ExecContext(ctx, `DELETE FROM channel_models WHERE id = ?`, id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
n, _ := res.RowsAffected()
|
|
if n == 0 {
|
|
return sql.ErrNoRows
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *ChannelStore) UserOwns(ctx context.Context, channelID, userID string) (bool, error) {
|
|
var exists bool
|
|
err := DB.QueryRowContext(ctx,
|
|
"SELECT EXISTS(SELECT 1 FROM channels WHERE id = ? AND user_id = ?)",
|
|
channelID, userID).Scan(&exists)
|
|
return exists, err
|
|
}
|
|
|
|
// ResolveWorkspaceID returns the effective workspace for a channel.
|
|
// Resolution: channel.workspace_id > project.workspace_id.
|
|
// Returns empty string if no workspace is bound.
|
|
func (s *ChannelStore) ResolveWorkspaceID(ctx context.Context, channelID string) (string, error) {
|
|
var wsID sql.NullString
|
|
err := DB.QueryRowContext(ctx, `
|
|
SELECT COALESCE(c.workspace_id, p.workspace_id)
|
|
FROM channels c
|
|
LEFT JOIN project_channels pc ON pc.channel_id = c.id
|
|
LEFT JOIN projects p ON p.id = pc.project_id
|
|
WHERE c.id = ?
|
|
LIMIT 1`, channelID).Scan(&wsID)
|
|
if err != nil {
|
|
if err == sql.ErrNoRows {
|
|
return "", nil
|
|
}
|
|
return "", err
|
|
}
|
|
return NullableString(wsID), nil
|
|
}
|
|
|
|
// ── Channel Participants (ICD §3.7) ──────────
|
|
|
|
func (s *ChannelStore) AddParticipant(ctx context.Context, p *models.ChannelParticipant) error {
|
|
_, err := DB.ExecContext(ctx, `
|
|
INSERT INTO channel_participants (id, channel_id, participant_type, participant_id, role, display_name, avatar_url)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)`,
|
|
store.NewID(), p.ChannelID, p.ParticipantType, p.ParticipantID, p.Role, p.DisplayName, p.AvatarURL)
|
|
return err
|
|
}
|
|
|
|
func (s *ChannelStore) ListParticipants(ctx context.Context, channelID string) ([]models.ChannelParticipant, error) {
|
|
rows, err := DB.QueryContext(ctx, `
|
|
SELECT cp.id, cp.channel_id, cp.participant_type, cp.participant_id, cp.role,
|
|
COALESCE(NULLIF(cp.display_name,''), NULLIF(u.display_name,''), u.username) AS display_name,
|
|
COALESCE(NULLIF(cp.avatar_url,''), u.avatar_url) AS avatar_url,
|
|
cp.joined_at
|
|
FROM channel_participants cp
|
|
LEFT JOIN users u ON cp.participant_type = 'user' AND cp.participant_id = u.id
|
|
WHERE cp.channel_id = ? ORDER BY cp.joined_at`, channelID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var result []models.ChannelParticipant
|
|
for rows.Next() {
|
|
var p models.ChannelParticipant
|
|
var joinedStr string
|
|
if err := rows.Scan(&p.ID, &p.ChannelID, &p.ParticipantType, &p.ParticipantID,
|
|
&p.Role, &p.DisplayName, &p.AvatarURL, &joinedStr); err != nil {
|
|
return nil, err
|
|
}
|
|
p.JoinedAt, _ = time.Parse("2006-01-02 15:04:05", joinedStr)
|
|
result = append(result, p)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
func (s *ChannelStore) GetParticipantByID(ctx context.Context, id string) (*models.ChannelParticipant, error) {
|
|
var p models.ChannelParticipant
|
|
var joinedStr string
|
|
err := DB.QueryRowContext(ctx, `
|
|
SELECT id, channel_id, participant_type, participant_id, role,
|
|
display_name, avatar_url, joined_at
|
|
FROM channel_participants WHERE id = ?`, id).Scan(
|
|
&p.ID, &p.ChannelID, &p.ParticipantType, &p.ParticipantID,
|
|
&p.Role, &p.DisplayName, &p.AvatarURL, &joinedStr)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
p.JoinedAt, _ = time.Parse("2006-01-02 15:04:05", joinedStr)
|
|
return &p, nil
|
|
}
|
|
|
|
func (s *ChannelStore) UpdateParticipantRole(ctx context.Context, id, role string) error {
|
|
_, err := DB.ExecContext(ctx, `UPDATE channel_participants SET role = ? WHERE id = ?`, role, id)
|
|
return err
|
|
}
|
|
|
|
func (s *ChannelStore) RemoveParticipant(ctx context.Context, id string) error {
|
|
res, err := DB.ExecContext(ctx, `DELETE FROM channel_participants WHERE id = ?`, id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
n, _ := res.RowsAffected()
|
|
if n == 0 {
|
|
return sql.ErrNoRows
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *ChannelStore) IsParticipant(ctx context.Context, channelID, pType, pID string) (bool, error) {
|
|
var count int
|
|
err := DB.QueryRowContext(ctx, `
|
|
SELECT COUNT(*) FROM channel_participants
|
|
WHERE channel_id = ? AND participant_type = ? AND participant_id = ?`,
|
|
channelID, pType, pID).Scan(&count)
|
|
return count > 0, err
|
|
}
|
|
|
|
func (s *ChannelStore) GetParticipantRole(ctx context.Context, channelID, pType, pID string) (string, error) {
|
|
var role string
|
|
err := DB.QueryRowContext(ctx, `
|
|
SELECT role FROM channel_participants
|
|
WHERE channel_id = ? AND participant_type = ? AND participant_id = ?`,
|
|
channelID, pType, pID).Scan(&role)
|
|
return role, err
|
|
}
|
|
|
|
func (s *ChannelStore) CountParticipantsByRole(ctx context.Context, channelID, role string) (int, error) {
|
|
var count int
|
|
err := DB.QueryRowContext(ctx, `
|
|
SELECT COUNT(*) FROM channel_participants WHERE channel_id = ? AND role = ?`,
|
|
channelID, role).Scan(&count)
|
|
return count, err
|
|
}
|
|
|
|
func (s *ChannelStore) DeleteModelByPersona(ctx context.Context, channelID, personaID string) error {
|
|
_, err := DB.ExecContext(ctx,
|
|
`DELETE FROM channel_models WHERE channel_id = ? AND persona_id = ?`,
|
|
channelID, personaID)
|
|
return err
|
|
}
|
|
|
|
// ── Admin: Archived Channel Management ──────
|
|
|
|
func (s *ChannelStore) ListArchived(ctx context.Context, opts store.ListOptions) ([]store.ArchivedChannel, int, error) {
|
|
// Count
|
|
var total int
|
|
DB.QueryRowContext(ctx,
|
|
`SELECT COUNT(*) FROM channels WHERE is_archived = 1`).Scan(&total)
|
|
|
|
rows, err := DB.QueryContext(ctx, `
|
|
SELECT c.id, c.title, c.type, COALESCE(u.username, '') AS owner_name,
|
|
c.updated_at,
|
|
(SELECT COUNT(*) FROM messages WHERE channel_id = c.id) AS message_count
|
|
FROM channels c
|
|
LEFT JOIN users u ON u.id = c.user_id
|
|
WHERE c.is_archived = 1
|
|
ORDER BY c.updated_at DESC
|
|
LIMIT ? OFFSET ?`, opts.Limit, opts.Offset)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var channels []store.ArchivedChannel
|
|
for rows.Next() {
|
|
var ch store.ArchivedChannel
|
|
if rows.Scan(&ch.ID, &ch.Title, &ch.Type, &ch.OwnerName, &ch.UpdatedAt, &ch.MessageCount) == nil {
|
|
channels = append(channels, ch)
|
|
}
|
|
}
|
|
if channels == nil {
|
|
channels = []store.ArchivedChannel{}
|
|
}
|
|
return channels, total, rows.Err()
|
|
}
|
|
|
|
func (s *ChannelStore) Purge(ctx context.Context, id string) error {
|
|
// Verify the channel is actually archived
|
|
var isArchived int
|
|
err := DB.QueryRowContext(ctx,
|
|
`SELECT is_archived FROM channels WHERE id = ?`, id).Scan(&isArchived)
|
|
if err != nil {
|
|
return fmt.Errorf("channel not found")
|
|
}
|
|
if isArchived == 0 {
|
|
return fmt.Errorf("channel must be archived before purging")
|
|
}
|
|
|
|
// Hard delete (CASCADE removes messages, participants, models)
|
|
_, err = DB.ExecContext(ctx, `DELETE FROM channels WHERE id = ?`, id)
|
|
return err
|
|
}
|
|
// ── CS1 additions (v0.29.0) ─────────────────────────────────────────────
|
|
|
|
func (s *ChannelStore) FindExistingDM(ctx context.Context, userID1, userID2 string) (string, error) {
|
|
var channelID string
|
|
err := DB.QueryRowContext(ctx, `
|
|
SELECT cp1.channel_id FROM channel_participants cp1
|
|
JOIN channel_participants cp2 ON cp1.channel_id = cp2.channel_id
|
|
JOIN channels c ON c.id = cp1.channel_id
|
|
WHERE c.type = 'dm'
|
|
AND cp1.participant_type = 'user' AND cp1.participant_id = ?
|
|
AND cp2.participant_type = 'user' AND cp2.participant_id = ?
|
|
LIMIT 1
|
|
`, userID1, userID2).Scan(&channelID)
|
|
if err == sql.ErrNoRows {
|
|
return "", nil
|
|
}
|
|
return channelID, err
|
|
}
|
|
|
|
func (s *ChannelStore) GetUnreadCount(ctx context.Context, channelID, userID string) (int, error) {
|
|
var count int
|
|
err := DB.QueryRowContext(ctx, `
|
|
SELECT COUNT(*) FROM messages m
|
|
JOIN channel_participants cp ON cp.channel_id = m.channel_id
|
|
WHERE cp.channel_id = ?
|
|
AND cp.participant_type = 'user' AND cp.participant_id = ?
|
|
AND m.created_at > cp.last_read_at
|
|
`, channelID, userID).Scan(&count)
|
|
return count, err
|
|
}
|
|
|
|
func (s *ChannelStore) DeleteByOwner(ctx context.Context, channelID, userID string) (int64, error) {
|
|
result, err := DB.ExecContext(ctx,
|
|
`DELETE FROM channels WHERE id = ? AND user_id = ?`,
|
|
channelID, userID)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return result.RowsAffected()
|
|
}
|
|
|
|
func (s *ChannelStore) ArchiveForRetention(ctx context.Context, channelID string, purgeAfter time.Time) error {
|
|
_, err := DB.ExecContext(ctx, `
|
|
UPDATE channels SET is_archived = 1, purge_after = ?, updated_at = datetime('now')
|
|
WHERE id = ?
|
|
`, purgeAfter.UTC().Format("2006-01-02T15:04:05Z"), channelID)
|
|
return err
|
|
}
|
|
|
|
func (s *ChannelStore) ListPurgeable(ctx context.Context) ([]string, error) {
|
|
rows, err := DB.QueryContext(ctx, `
|
|
SELECT id FROM channels WHERE purge_after IS NOT NULL AND purge_after <= datetime('now')
|
|
`)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var ids []string
|
|
for rows.Next() {
|
|
var id string
|
|
if rows.Scan(&id) == nil {
|
|
ids = append(ids, id)
|
|
}
|
|
}
|
|
return ids, rows.Err()
|
|
}
|
|
|
|
func (s *ChannelStore) MarkRead(ctx context.Context, channelID, userID string) error {
|
|
_, err := DB.ExecContext(ctx, `
|
|
UPDATE channel_participants
|
|
SET last_read_at = datetime('now')
|
|
WHERE channel_id = ? AND participant_type = 'user' AND participant_id = ?
|
|
`, channelID, userID)
|
|
if err != nil {
|
|
return nil
|
|
}
|
|
|
|
var latestMsgID *string
|
|
_ = DB.QueryRowContext(ctx, `
|
|
SELECT id FROM messages WHERE channel_id = ? ORDER BY created_at DESC LIMIT 1
|
|
`, channelID).Scan(&latestMsgID)
|
|
if latestMsgID != nil {
|
|
_, _ = DB.ExecContext(ctx, `
|
|
UPDATE channel_participants
|
|
SET last_read_message_id = ?
|
|
WHERE channel_id = ? AND participant_type = 'user' AND participant_id = ?
|
|
`, *latestMsgID, channelID, userID)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (s *ChannelStore) CountParticipantsByType(ctx context.Context, channelID, pType string) (int, error) {
|
|
var count int
|
|
err := DB.QueryRowContext(ctx, `
|
|
SELECT COUNT(*) FROM channel_participants
|
|
WHERE channel_id = ? AND participant_type = ?
|
|
`, channelID, pType).Scan(&count)
|
|
return count, err
|
|
}
|
|
|
|
// ── CS2 additions (v0.29.0) ─────────────────────────────────────────────
|
|
|
|
func (s *ChannelStore) CountAll(ctx context.Context) (int, error) {
|
|
var count int
|
|
err := DB.QueryRowContext(ctx, "SELECT COUNT(*) FROM channels").Scan(&count)
|
|
return count, err
|
|
}
|
|
|
|
// ── Workflow instance state (v0.29.0-cs3) ───────────────────────────────
|
|
|
|
func (s *ChannelStore) SetWorkflowInstance(ctx context.Context, channelID, workflowID string, version int, stageData json.RawMessage, status string) error {
|
|
now := time.Now().UTC().Format(timeFmt)
|
|
_, err := DB.ExecContext(ctx, `
|
|
UPDATE channels
|
|
SET workflow_id = ?, workflow_version = ?, current_stage = 0,
|
|
stage_data = ?, workflow_status = ?, last_activity_at = ?,
|
|
stage_entered_at = ?
|
|
WHERE id = ?
|
|
`, workflowID, version, stageData, status, now, now, channelID)
|
|
return err
|
|
}
|
|
|
|
func (s *ChannelStore) GetWorkflowStatus(ctx context.Context, channelID string) (*store.WorkflowChannelStatus, error) {
|
|
var ws store.WorkflowChannelStatus
|
|
var stageData []byte
|
|
err := DB.QueryRowContext(ctx, `
|
|
SELECT workflow_id, workflow_version, current_stage,
|
|
COALESCE(stage_data, '{}'), COALESCE(workflow_status, 'active'),
|
|
last_activity_at, stage_entered_at
|
|
FROM channels WHERE id = ? AND type = 'workflow'
|
|
`, channelID).Scan(&ws.WorkflowID, &ws.WorkflowVersion,
|
|
&ws.CurrentStage, &stageData, &ws.Status, &ws.LastActivityAt, &ws.StageEnteredAt)
|
|
if err == sql.ErrNoRows {
|
|
return nil, nil
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
ws.StageData = stageData
|
|
return &ws, nil
|
|
}
|
|
|
|
func (s *ChannelStore) AdvanceWorkflowStage(ctx context.Context, channelID string, nextStage int, stageData json.RawMessage) error {
|
|
now := time.Now().UTC().Format(timeFmt)
|
|
_, err := DB.ExecContext(ctx, `
|
|
UPDATE channels
|
|
SET current_stage = ?, stage_data = ?, last_activity_at = ?,
|
|
stage_entered_at = ?
|
|
WHERE id = ?
|
|
`, nextStage, stageData, now, now, channelID)
|
|
return err
|
|
}
|
|
|
|
func (s *ChannelStore) CompleteWorkflow(ctx context.Context, channelID string, finalStage int, stageData json.RawMessage) error {
|
|
_, err := DB.ExecContext(ctx, `
|
|
UPDATE channels
|
|
SET current_stage = ?, workflow_status = 'completed',
|
|
stage_data = ?, last_activity_at = ?, ai_mode = 'off'
|
|
WHERE id = ?
|
|
`, finalStage, stageData, time.Now().UTC().Format(timeFmt), channelID)
|
|
return err
|
|
}
|
|
|
|
func (s *ChannelStore) CancelWorkflow(ctx context.Context, channelID string) error {
|
|
_, err := DB.ExecContext(ctx, `
|
|
UPDATE channels
|
|
SET workflow_status = 'cancelled', ai_mode = 'off', last_activity_at = ?
|
|
WHERE id = ? AND type = 'workflow'
|
|
`, time.Now().UTC().Format(timeFmt), channelID)
|
|
return err
|
|
}
|
|
|
|
func (s *ChannelStore) RejectWorkflowToStage(ctx context.Context, channelID string, stage int) error {
|
|
now := time.Now().UTC().Format(timeFmt)
|
|
_, err := DB.ExecContext(ctx, `
|
|
UPDATE channels SET current_stage = ?, last_activity_at = ?,
|
|
stage_entered_at = ? WHERE id = ?
|
|
`, stage, now, now, channelID)
|
|
return err
|
|
}
|
|
|
|
func (s *ChannelStore) GetStageData(ctx context.Context, channelID string) (json.RawMessage, error) {
|
|
var data json.RawMessage
|
|
err := DB.QueryRowContext(ctx, `
|
|
SELECT COALESCE(stage_data, '{}') FROM channels WHERE id = ?
|
|
`, channelID).Scan(&data)
|
|
return data, err
|
|
}
|
|
|
|
// ── Background job helpers (v0.29.0-cs4) ────────────────────────────────
|
|
|
|
func (s *ChannelStore) MarkStaleWorkflows(ctx context.Context, cutoff time.Time) (int64, error) {
|
|
result, err := DB.ExecContext(ctx, `
|
|
UPDATE channels
|
|
SET workflow_status = 'stale'
|
|
WHERE type = 'workflow'
|
|
AND workflow_status = 'active'
|
|
AND last_activity_at < ?
|
|
`, cutoff.Format(timeFmt))
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return result.RowsAffected()
|
|
}
|
|
|
|
func (s *ChannelStore) EnforceWorkflowRetention(ctx context.Context) (int64, error) {
|
|
// SQLite doesn't support JSON operators for retention policy evaluation.
|
|
// This is a no-op on SQLite; retention enforcement is PG-only.
|
|
return 0, nil
|
|
}
|
|
|
|
func (s *ChannelStore) GetTypeAndAllowAnonymous(ctx context.Context, channelID string) (string, bool, error) {
|
|
var chType string
|
|
var allowAnon bool
|
|
err := DB.QueryRowContext(ctx,
|
|
`SELECT type, allow_anonymous FROM channels WHERE id = ?`, channelID).Scan(&chType, &allowAnon)
|
|
return chType, allowAnon, err
|
|
}
|
|
|
|
// ── CS5b additions (v0.29.0) ────────────────────────────────────────────
|
|
|
|
func (s *ChannelStore) FindCompactionCandidates(ctx context.Context, activityBefore, createdAfter time.Time, minMessages, minChars, limit int) ([]models.Channel, error) {
|
|
// Auto-compaction scanner is PG-only (uses settings::text cast).
|
|
return []models.Channel{}, nil
|
|
}
|
|
|
|
// ── CS7a additions (v0.29.0) ────────────────────────────────────────────
|
|
|
|
func (s *ChannelStore) GetProviderConfigID(ctx context.Context, channelID string) (*string, error) {
|
|
var configID sql.NullString
|
|
err := DB.QueryRowContext(ctx,
|
|
`SELECT provider_config_id FROM channels WHERE id = ?`, channelID).Scan(&configID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return NullableStringPtr(configID), nil
|
|
}
|
|
|
|
func (s *ChannelStore) UserCanAccess(ctx context.Context, channelID, userID string) (bool, error) {
|
|
var ok bool
|
|
err := DB.QueryRowContext(ctx, `
|
|
SELECT EXISTS(
|
|
SELECT 1 FROM channels WHERE id = ? AND user_id = ?
|
|
UNION ALL
|
|
SELECT 1 FROM channel_participants
|
|
WHERE channel_id = ? AND participant_type = 'user' AND participant_id = ?
|
|
LIMIT 1
|
|
)`, channelID, userID, channelID, userID).Scan(&ok)
|
|
return ok, err
|
|
}
|
|
|
|
// ── CS7b additions (v0.29.0) ────────────────────────────────────────────
|
|
|
|
const channelListCols = `c.id, c.user_id, c.title, c.type, c.ai_mode, c.topic,
|
|
c.description, c.model, c.provider_config_id,
|
|
c.system_prompt, c.is_archived, c.is_pinned, c.folder, c.folder_id, c.project_id, c.workspace_id,
|
|
c.tags, c.settings,
|
|
COALESCE(mc.cnt, 0) AS message_count,
|
|
c.created_at, c.updated_at`
|
|
|
|
const channelListFrom = `channels c
|
|
LEFT JOIN (
|
|
SELECT channel_id, COUNT(*) AS cnt FROM messages WHERE deleted_at IS NULL GROUP BY channel_id
|
|
) mc ON mc.channel_id = c.id`
|
|
|
|
func (s *ChannelStore) ListFiltered(ctx context.Context, userID string, f store.ChannelListFilter) ([]store.ChannelListItem, int, error) {
|
|
b := NewSelect(channelListCols, channelListFrom)
|
|
b.Where("(c.user_id = ? OR c.id IN (SELECT channel_id FROM channel_participants WHERE participant_type = 'user' AND participant_id = ?))", userID, userID)
|
|
archivedVal := 0
|
|
if f.Archived {
|
|
archivedVal = 1
|
|
}
|
|
b.Where("c.is_archived = ?", archivedVal)
|
|
|
|
if len(f.Types) == 1 {
|
|
b.Where("c.type = ?", f.Types[0])
|
|
} else if len(f.Types) > 1 {
|
|
placeholders := make([]string, len(f.Types))
|
|
for i, t := range f.Types {
|
|
placeholders[i] = "?"
|
|
b.args = append(b.args, t)
|
|
}
|
|
b.WhereRaw("c.type IN (" + strings.Join(placeholders, ",") + ")")
|
|
}
|
|
if f.Folder != "" {
|
|
b.Where("c.folder = ?", f.Folder)
|
|
}
|
|
if f.FolderID != "" {
|
|
b.Where("c.folder_id = ?", f.FolderID)
|
|
}
|
|
if f.Search != "" {
|
|
b.Where("c.title LIKE ?", "%"+f.Search+"%")
|
|
}
|
|
if f.ProjectID == "none" {
|
|
b.WhereRaw("c.project_id IS NULL")
|
|
} else if f.ProjectID != "" {
|
|
b.Where("c.project_id = ?", f.ProjectID)
|
|
}
|
|
|
|
// Count
|
|
countQ, countArgs := b.CountBuild()
|
|
var total int
|
|
DB.QueryRowContext(ctx, countQ, countArgs...).Scan(&total)
|
|
|
|
b.OrderBy("c.is_pinned DESC, c.updated_at", "DESC")
|
|
b.Paginate(f.ListOptions)
|
|
|
|
q, args := b.Build()
|
|
rows, err := DB.QueryContext(ctx, q, args...)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
items, err := scanChannelListItems(rows)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
|
|
// Compute unread counts
|
|
for i := range items {
|
|
DB.QueryRowContext(ctx, `
|
|
SELECT COUNT(*) FROM messages m
|
|
JOIN channel_participants cp ON cp.channel_id = m.channel_id
|
|
WHERE cp.channel_id = ?
|
|
AND cp.participant_type = 'user' AND cp.participant_id = ?
|
|
AND m.created_at > cp.last_read_at
|
|
AND m.deleted_at IS NULL
|
|
`, items[i].ID, userID).Scan(&items[i].UnreadCount)
|
|
}
|
|
|
|
return items, total, nil
|
|
}
|
|
|
|
func (s *ChannelStore) GetForUser(ctx context.Context, channelID, userID string) (*store.ChannelListItem, error) {
|
|
row := DB.QueryRowContext(ctx, fmt.Sprintf(`
|
|
SELECT %s FROM %s
|
|
WHERE c.id = ? AND (c.user_id = ? OR c.id IN (
|
|
SELECT channel_id FROM channel_participants WHERE participant_type = 'user' AND participant_id = ?
|
|
))
|
|
`, channelListCols, channelListFrom), channelID, userID, userID)
|
|
|
|
var item store.ChannelListItem
|
|
var tags, settings string
|
|
err := row.Scan(
|
|
&item.ID, &item.UserID, &item.Title, &item.Type, &item.AiMode, &item.Topic,
|
|
&item.Description, &item.Model, &item.ProviderConfigID,
|
|
&item.SystemPrompt, &item.IsArchived, &item.IsPinned, &item.Folder, &item.FolderID, &item.ProjectID, &item.WorkspaceID,
|
|
&tags, &settings,
|
|
&item.MessageCount, st(&item.CreatedAtTime), st(&item.UpdatedAtTime),
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
item.Tags = ScanArray(tags)
|
|
if item.Tags == nil {
|
|
item.Tags = []string{}
|
|
}
|
|
item.Settings = safeJSONString(settings)
|
|
item.CreatedAt = item.CreatedAtTime.Format("2006-01-02T15:04:05Z")
|
|
item.UpdatedAt = item.UpdatedAtTime.Format("2006-01-02T15:04:05Z")
|
|
return &item, nil
|
|
}
|
|
|
|
func (s *ChannelStore) CreateFull(ctx context.Context, ch *models.Channel, folder string, tags []string, aiMode string,
|
|
ownerUserID string, dmPartnerIDs []string, defaultModel, defaultConfigID string) error {
|
|
ch.ID = store.NewID()
|
|
now := time.Now().UTC()
|
|
ch.CreatedAt = now
|
|
ch.UpdatedAt = now
|
|
|
|
if tags == nil {
|
|
tags = []string{}
|
|
}
|
|
|
|
tx, err := DB.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
tagsJSON := ArrayToJSON(tags)
|
|
_, err = tx.ExecContext(ctx, `
|
|
INSERT INTO channels (id, user_id, title, type, description, model, system_prompt,
|
|
provider_config_id, folder, folder_id, tags, ai_mode, created_at, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
|
ch.ID, ch.UserID, ch.Title, ch.Type, ch.Description, ch.Model, ch.SystemPrompt,
|
|
models.NullString(ch.ProviderConfigID), folder, models.NullString(ch.FolderID),
|
|
tagsJSON, aiMode, now.Format(timeFmt), now.Format(timeFmt),
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("CreateFull insert channel: %w", err)
|
|
}
|
|
|
|
// Add owner participant
|
|
_, _ = tx.ExecContext(ctx, `
|
|
INSERT INTO channel_participants (id, channel_id, participant_type, participant_id, role)
|
|
VALUES (?, ?, 'user', ?, 'owner')
|
|
ON CONFLICT DO NOTHING`, store.NewID(), ch.ID, ownerUserID)
|
|
|
|
// Add DM partner participants
|
|
for _, pid := range dmPartnerIDs {
|
|
if pid == ownerUserID {
|
|
continue
|
|
}
|
|
_, _ = tx.ExecContext(ctx, `
|
|
INSERT INTO channel_participants (id, channel_id, participant_type, participant_id, role)
|
|
VALUES (?, ?, 'user', ?, 'member')
|
|
ON CONFLICT DO NOTHING`, store.NewID(), ch.ID, pid)
|
|
}
|
|
|
|
// Auto-create channel_model if model specified
|
|
if defaultModel != "" {
|
|
_, _ = tx.ExecContext(ctx, `
|
|
INSERT INTO channel_models (id, channel_id, model_id, provider_config_id, is_default)
|
|
VALUES (?, ?, ?, ?, 1)
|
|
ON CONFLICT DO NOTHING`, store.NewID(), ch.ID, defaultModel, models.NullString(&defaultConfigID))
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|
|
|
|
func (s *ChannelStore) MergeSettings(ctx context.Context, channelID string, settingsJSON json.RawMessage) error {
|
|
_, err := DB.ExecContext(ctx,
|
|
`UPDATE channels SET settings = json_patch(COALESCE(settings, '{}'), ?) WHERE id = ?`,
|
|
string(settingsJSON), channelID)
|
|
return err
|
|
}
|
|
|
|
// ── Monitoring (v0.35.0) ────────────────────
|
|
|
|
func (s *ChannelStore) ListByType(ctx context.Context, channelType string) ([]models.Channel, error) {
|
|
rows, err := DB.QueryContext(ctx, `
|
|
SELECT id, user_id, title, COALESCE(description, ''), type,
|
|
team_id, created_at
|
|
FROM channels WHERE type = ?
|
|
ORDER BY created_at DESC`, channelType)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var result []models.Channel
|
|
for rows.Next() {
|
|
var ch models.Channel
|
|
if err := rows.Scan(&ch.ID, &ch.UserID, &ch.Title, &ch.Description,
|
|
&ch.Type, &ch.TeamID, st(&ch.CreatedAt)); err != nil {
|
|
return nil, err
|
|
}
|
|
result = append(result, ch)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
// ── CS7b helpers ────────────────────────────
|
|
|
|
func scanChannelListItems(rows *sql.Rows) ([]store.ChannelListItem, error) {
|
|
var result []store.ChannelListItem
|
|
for rows.Next() {
|
|
var item store.ChannelListItem
|
|
var tags, settings string
|
|
err := rows.Scan(
|
|
&item.ID, &item.UserID, &item.Title, &item.Type, &item.AiMode, &item.Topic,
|
|
&item.Description, &item.Model, &item.ProviderConfigID,
|
|
&item.SystemPrompt, &item.IsArchived, &item.IsPinned, &item.Folder, &item.FolderID, &item.ProjectID, &item.WorkspaceID,
|
|
&tags, &settings,
|
|
&item.MessageCount, st(&item.CreatedAtTime), st(&item.UpdatedAtTime),
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
item.Tags = ScanArray(tags)
|
|
if item.Tags == nil {
|
|
item.Tags = []string{}
|
|
}
|
|
item.Settings = safeJSONString(settings)
|
|
item.CreatedAt = item.CreatedAtTime.Format("2006-01-02T15:04:05Z")
|
|
item.UpdatedAt = item.UpdatedAtTime.Format("2006-01-02T15:04:05Z")
|
|
result = append(result, item)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
func safeJSONString(s string) json.RawMessage {
|
|
if s == "" || s == "null" {
|
|
return json.RawMessage("{}")
|
|
}
|
|
return json.RawMessage(s)
|
|
}
|