Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
6.7 kB · 177 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178package model
import ( "context" "errors" "fmt" "hash/fnv" "strings" "time"
"github.com/rivo/uniseg" glex "github.com/streamplace/glex/runtime" "gorm.io/gorm" "stream.place/streamplace/pkg/appbsky" "stream.place/streamplace/pkg/placestream")
type ChatMessage struct { CID string `json:"cid" gorm:"primaryKey;column:cid"` URI string `json:"uri" gorm:"column:uri"` CreatedAt time.Time `json:"createdAt" gorm:"column:created_at;index:idx_recent_messages,priority:2"` ChatMessage *[]byte `json:"chatMessage" gorm:"column:chat_message"` RepoDID string `json:"repoDID" gorm:"column:repo_did"` Repo *Repo `json:"repo,omitempty" gorm:"foreignKey:DID;references:RepoDID"` ChatProfile *ChatProfile `json:"chatProfile,omitempty" gorm:"foreignKey:RepoDID;references:RepoDID"` IndexedAt *time.Time `json:"indexedAt,omitempty" gorm:"column:indexed_at"` StreamerRepoDID string `json:"streamerRepoDID" gorm:"column:streamer_repo_did;idx_recent_messages,priority:1"` StreamerRepo *Repo `json:"streamerRepo,omitempty" gorm:"foreignKey:DID;references:StreamerRepoDID"` ReplyToCID *string `json:"replyToCID,omitempty" gorm:"column:reply_to_cid"` ReplyTo *ChatMessage `json:"replyTo,omitempty" gorm:"foreignKey:ReplyToCID;references:CID"` DeletedAt *time.Time `json:"deletedAt,omitempty" gorm:"column:deleted_at"`}
// hashString creates a hash from a string, used for deterministic color selectionfunc hashString(s string) int { h := fnv.New32a() h.Write([]byte(s)) return int(h.Sum32())}
func (m *ChatMessage) ToStreamplaceMessageView() (*placestream.ChatDefs_MessageView, error) { var msg placestream.ChatMessage if err := glex.DecodeCBOR(*m.ChatMessage, &msg); err != nil { return nil, fmt.Errorf("error decoding chat message: %w", err) } // Truncate overlong message text if uniseg.GraphemeClusterCount(msg.Text) > 300 { gr := uniseg.NewGraphemes(msg.Text) var result strings.Builder for count := 0; count < 300 && gr.Next(); count++ { result.WriteString(gr.Str()) } msg.Text = result.String() }
message := placestream.ChatDefs_MessageView{ LexiconTypeID: "place.stream.chat.defs#messageView", } message.Uri = m.URI message.Cid = m.CID message.Author = appbsky.ActorDefs_ProfileViewBasic{ Did: m.RepoDID, } if m.Repo != nil { message.Author.Handle = m.Repo.Handle } message.Record = &glex.LexiconTypeDecoder{Val: &msg} message.IndexedAt = m.IndexedAt.UTC().Format(time.RFC3339Nano) if m.ChatProfile != nil { scp, err := m.ChatProfile.ToStreamplaceChatProfile() if err != nil { return nil, fmt.Errorf("error converting chat profile to streamplace chat profile: %w", err) } message.ChatProfile = &scp } else { // If no chat profile exists, create a default one with a color based on the user's DID defaultColor := DefaultColors[hashString(m.RepoDID)%len(DefaultColors)] message.ChatProfile = &placestream.ChatProfile{ Color: &defaultColor, }
} if m.ReplyTo != nil { replyTo, err := m.ReplyTo.ToStreamplaceMessageView() if err != nil { return nil, fmt.Errorf("error converting reply to to streamplace message view: %w", err) } message.ReplyTo = &placestream.ChatDefs_MessageView_ReplyTo{ ChatDefs_MessageView: replyTo, } } return &message, nil}
// CreateChatMessage indexes one chat message. The table is keyed by record CID,// so every conflict here is a redelivery of a message we already have; an edited// message is a different CID and lands as its own row, as it always has.func (m *DBModel) CreateChatMessage(ctx context.Context, message *ChatMessage) error { return createOrVerify(ctx, m, message, map[string]any{"cid": message.CID})}
func (m *DBModel) DeleteChatMessage(ctx context.Context, uri string, deletedAt *time.Time) error { tx := m.DB.Model(&ChatMessage{}).Where("uri = ?", uri).Update("deleted_at", deletedAt) if tx.Error != nil { return tx.Error } if tx.RowsAffected == 0 { return fmt.Errorf("no chat message found for uri: %s", uri) } return nil}
func (m *DBModel) GetChatMessage(uri string) (*ChatMessage, error) { var message ChatMessage err := m.DB. Preload("Repo"). Preload("ChatProfile"). Preload("ReplyTo"). Preload("ReplyTo.Repo"). Preload("ReplyTo.ChatProfile"). Where("uri = ?", uri). Where("deleted_at IS NULL"). First(&message). Error if errors.Is(err, gorm.ErrRecordNotFound) { return nil, nil } if err != nil { return nil, fmt.Errorf("error retrieving chat message: %w", err) } return &message, nil}
func (m *DBModel) MostRecentChatMessages(repoDID string, since time.Time) ([]placestream.ChatDefs_MessageView, error) { dbmessages := []ChatMessage{} query := m.DB. Preload("Repo"). Preload("ChatProfile"). Preload("ReplyTo"). Preload("ReplyTo.Repo"). Preload("ReplyTo.ChatProfile"). Where("streamer_repo_did = ?", repoDID) if !since.IsZero() { // Chat is a live conversation, not an archive: withhold history older // than the node's retention window so a viewer arriving later lands on // an empty chat instead of a wall of stale messages. A zero `since` // serves everything, which is what a 0 retention asks for. query = query.Where("chat_messages.created_at >= ?", since) } err := query. // Exclude messages from users blocked by the streamer Joins("LEFT JOIN blocks ON blocks.repo_did = chat_messages.streamer_repo_did AND blocks.subject_did = chat_messages.repo_did"). Where("blocks.rkey IS NULL"). // Only include messages where no block exists // Exclude gated messages Joins("LEFT JOIN gates ON gates.repo_did = chat_messages.streamer_repo_did AND gates.hidden_message = chat_messages.uri"). Where("gates.hidden_message IS NULL"). // Only include messages where no gate exists // Exclude labeled messages Joins("LEFT JOIN labels ON labels.uri = chat_messages.uri"). Where("labels.uri IS NULL"). // Only include messages where no label exists // Exclude deleted messages Where("chat_messages.deleted_at IS NULL"). Limit(100). Order("chat_messages.created_at DESC"). Find(&dbmessages).Error if err != nil { return nil, fmt.Errorf("error retrieving replies: %w", err) } spmessages := []placestream.ChatDefs_MessageView{} for _, m := range dbmessages { spmessage, err := m.ToStreamplaceMessageView() if err != nil { return nil, fmt.Errorf("error converting feed post to bsky post view: %w", err) } spmessages = append(spmessages, *spmessage) } return spmessages, nil}