|
1
|
// Package chathistory provides a synchronous wrapper around the IRCv3 |
|
2
|
// CHATHISTORY extension for use with girc clients. |
|
3
|
// |
|
4
|
// Usage: |
|
5
|
// |
|
6
|
// fetcher := chathistory.New(client) |
|
7
|
// msgs, err := fetcher.Latest(ctx, "#channel", 50) |
|
8
|
package chathistory |
|
9
|
|
|
10
|
import ( |
|
11
|
"context" |
|
12
|
"fmt" |
|
13
|
"strings" |
|
14
|
"sync" |
|
15
|
"time" |
|
16
|
|
|
17
|
"github.com/lrstanley/girc" |
|
18
|
) |
|
19
|
|
|
20
|
// Message is a single message returned by a CHATHISTORY query. |
|
21
|
type Message struct { |
|
22
|
At time.Time |
|
23
|
Nick string |
|
24
|
Account string |
|
25
|
Text string |
|
26
|
MsgID string |
|
27
|
} |
|
28
|
|
|
29
|
// Fetcher sends CHATHISTORY commands and collects the batched responses. |
|
30
|
type Fetcher struct { |
|
31
|
client *girc.Client |
|
32
|
|
|
33
|
mu sync.Mutex |
|
34
|
batches map[string]*batch // batchRef → accumulator |
|
35
|
waiters map[string]chan []Message // channel → result (one waiter per channel) |
|
36
|
handlers bool |
|
37
|
} |
|
38
|
|
|
39
|
type batch struct { |
|
40
|
channel string |
|
41
|
msgs []Message |
|
42
|
} |
|
43
|
|
|
44
|
// New creates a Fetcher and registers the necessary BATCH handlers on the |
|
45
|
// client. The client's Config.SupportedCaps should include |
|
46
|
// "draft/chathistory" (or "chathistory") so the capability is negotiated. |
|
47
|
func New(client *girc.Client) *Fetcher { |
|
48
|
f := &Fetcher{ |
|
49
|
client: client, |
|
50
|
batches: make(map[string]*batch), |
|
51
|
waiters: make(map[string]chan []Message), |
|
52
|
} |
|
53
|
f.registerHandlers() |
|
54
|
return f |
|
55
|
} |
|
56
|
|
|
57
|
func (f *Fetcher) registerHandlers() { |
|
58
|
f.mu.Lock() |
|
59
|
defer f.mu.Unlock() |
|
60
|
if f.handlers { |
|
61
|
return |
|
62
|
} |
|
63
|
f.handlers = true |
|
64
|
|
|
65
|
// BATCH open/close. |
|
66
|
f.client.Handlers.AddBg("BATCH", func(_ *girc.Client, e girc.Event) { |
|
67
|
if len(e.Params) < 1 { |
|
68
|
return |
|
69
|
} |
|
70
|
raw := e.Params[0] |
|
71
|
if strings.HasPrefix(raw, "+") { |
|
72
|
ref := raw[1:] |
|
73
|
if len(e.Params) >= 2 && e.Params[1] == "chathistory" { |
|
74
|
ch := "" |
|
75
|
if len(e.Params) >= 3 { |
|
76
|
ch = e.Params[2] |
|
77
|
} |
|
78
|
f.mu.Lock() |
|
79
|
f.batches[ref] = &batch{channel: ch} |
|
80
|
f.mu.Unlock() |
|
81
|
} |
|
82
|
} else if strings.HasPrefix(raw, "-") { |
|
83
|
ref := raw[1:] |
|
84
|
f.mu.Lock() |
|
85
|
b, ok := f.batches[ref] |
|
86
|
if ok { |
|
87
|
delete(f.batches, ref) |
|
88
|
if w, wok := f.waiters[b.channel]; wok { |
|
89
|
delete(f.waiters, b.channel) |
|
90
|
f.mu.Unlock() |
|
91
|
w <- b.msgs |
|
92
|
return |
|
93
|
} |
|
94
|
} |
|
95
|
f.mu.Unlock() |
|
96
|
} |
|
97
|
}) |
|
98
|
|
|
99
|
// Collect PRIVMSGs tagged with a tracked batch ref. |
|
100
|
f.client.Handlers.AddBg(girc.PRIVMSG, func(_ *girc.Client, e girc.Event) { |
|
101
|
batchRef, ok := e.Tags.Get("batch") |
|
102
|
if !ok || batchRef == "" { |
|
103
|
return |
|
104
|
} |
|
105
|
|
|
106
|
f.mu.Lock() |
|
107
|
b, tracked := f.batches[batchRef] |
|
108
|
if !tracked { |
|
109
|
f.mu.Unlock() |
|
110
|
return |
|
111
|
} |
|
112
|
|
|
113
|
nick := "" |
|
114
|
if e.Source != nil { |
|
115
|
nick = e.Source.Name |
|
116
|
} |
|
117
|
acct, _ := e.Tags.Get("account") |
|
118
|
msgID, _ := e.Tags.Get("msgid") |
|
119
|
|
|
120
|
b.msgs = append(b.msgs, Message{ |
|
121
|
At: e.Timestamp, |
|
122
|
Nick: nick, |
|
123
|
Account: acct, |
|
124
|
Text: e.Last(), |
|
125
|
MsgID: msgID, |
|
126
|
}) |
|
127
|
f.mu.Unlock() |
|
128
|
}) |
|
129
|
} |
|
130
|
|
|
131
|
// Latest fetches the N most recent messages from a channel using |
|
132
|
// CHATHISTORY LATEST. Blocks until the server responds or ctx expires. |
|
133
|
func (f *Fetcher) Latest(ctx context.Context, channel string, count int) ([]Message, error) { |
|
134
|
result := make(chan []Message, 1) |
|
135
|
|
|
136
|
f.mu.Lock() |
|
137
|
f.waiters[channel] = result |
|
138
|
f.mu.Unlock() |
|
139
|
|
|
140
|
if err := f.client.Cmd.SendRawf("CHATHISTORY LATEST %s * %d", channel, count); err != nil { |
|
141
|
f.mu.Lock() |
|
142
|
delete(f.waiters, channel) |
|
143
|
f.mu.Unlock() |
|
144
|
return nil, fmt.Errorf("chathistory: send: %w", err) |
|
145
|
} |
|
146
|
|
|
147
|
select { |
|
148
|
case msgs := <-result: |
|
149
|
return msgs, nil |
|
150
|
case <-ctx.Done(): |
|
151
|
f.mu.Lock() |
|
152
|
delete(f.waiters, channel) |
|
153
|
f.mu.Unlock() |
|
154
|
return nil, ctx.Err() |
|
155
|
} |
|
156
|
} |
|
157
|
|
|
158
|
// Before fetches up to count messages before the given timestamp. |
|
159
|
func (f *Fetcher) Before(ctx context.Context, channel string, before time.Time, count int) ([]Message, error) { |
|
160
|
result := make(chan []Message, 1) |
|
161
|
|
|
162
|
f.mu.Lock() |
|
163
|
f.waiters[channel] = result |
|
164
|
f.mu.Unlock() |
|
165
|
|
|
166
|
ts := before.UTC().Format("2006-01-02T15:04:05.000Z") |
|
167
|
if err := f.client.Cmd.SendRawf("CHATHISTORY BEFORE %s timestamp=%s %d", channel, ts, count); err != nil { |
|
168
|
f.mu.Lock() |
|
169
|
delete(f.waiters, channel) |
|
170
|
f.mu.Unlock() |
|
171
|
return nil, fmt.Errorf("chathistory: send: %w", err) |
|
172
|
} |
|
173
|
|
|
174
|
select { |
|
175
|
case msgs := <-result: |
|
176
|
return msgs, nil |
|
177
|
case <-ctx.Done(): |
|
178
|
f.mu.Lock() |
|
179
|
delete(f.waiters, channel) |
|
180
|
f.mu.Unlock() |
|
181
|
return nil, ctx.Err() |
|
182
|
} |
|
183
|
} |
|
184
|
|