1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
|
package message
import (
"context"
"encoding/json"
"github.com/cloudwego/eino/schema"
"github.com/google/uuid"
"github.com/kujtimiihoxha/termai/internal/db"
"github.com/kujtimiihoxha/termai/internal/pubsub"
)
type Message struct {
ID string
SessionID string
MessageData schema.Message
CreatedAt int64
UpdatedAt int64
}
type Service interface {
pubsub.Suscriber[Message]
Create(sessionID string, messageData schema.Message) (Message, error)
Get(id string) (Message, error)
List(sessionID string) ([]Message, error)
Delete(id string) error
DeleteSessionMessages(sessionID string) error
}
type service struct {
*pubsub.Broker[Message]
q db.Querier
ctx context.Context
}
func (s *service) Create(sessionID string, messageData schema.Message) (Message, error) {
messageDataJSON, err := json.Marshal(messageData)
if err != nil {
return Message{}, err
}
dbMessage, err := s.q.CreateMessage(s.ctx, db.CreateMessageParams{
ID: uuid.New().String(),
SessionID: sessionID,
MessageData: string(messageDataJSON),
})
if err != nil {
return Message{}, err
}
message := s.fromDBItem(dbMessage)
s.Publish(pubsub.CreatedEvent, message)
return message, nil
}
func (s *service) Delete(id string) error {
message, err := s.Get(id)
if err != nil {
return err
}
err = s.q.DeleteMessage(s.ctx, message.ID)
if err != nil {
return err
}
s.Publish(pubsub.DeletedEvent, message)
return nil
}
func (s *service) DeleteSessionMessages(sessionID string) error {
messages, err := s.List(sessionID)
if err != nil {
return err
}
for _, message := range messages {
if message.SessionID == sessionID {
err = s.Delete(message.ID)
if err != nil {
return err
}
}
}
return nil
}
func (s *service) Get(id string) (Message, error) {
dbMessage, err := s.q.GetMessage(s.ctx, id)
if err != nil {
return Message{}, err
}
return s.fromDBItem(dbMessage), nil
}
func (s *service) List(sessionID string) ([]Message, error) {
dbMessages, err := s.q.ListMessagesBySession(s.ctx, sessionID)
if err != nil {
return nil, err
}
messages := make([]Message, len(dbMessages))
for i, dbMessage := range dbMessages {
messages[i] = s.fromDBItem(dbMessage)
}
return messages, nil
}
func (s *service) fromDBItem(item db.Message) Message {
var messageData schema.Message
json.Unmarshal([]byte(item.MessageData), &messageData)
return Message{
ID: item.ID,
SessionID: item.SessionID,
MessageData: messageData,
CreatedAt: item.CreatedAt,
UpdatedAt: item.UpdatedAt,
}
}
func NewService(ctx context.Context, q db.Querier) Service {
return &service{
Broker: pubsub.NewBroker[Message](),
q: q,
ctx: ctx,
}
}
|
