1use std::fmt;
2
3use bytes::Bytes;
4use serde::Deserialize;
5use serde::Serialize;
6use ursula_shard::BucketStreamId;
7use ursula_shard::ShardPlacement;
8use ursula_stream::ColdChunkRef;
9use ursula_stream::ExternalPayloadRef;
10use ursula_stream::ProducerRequest;
11use ursula_stream::StreamSnapshot;
12
13use crate::request::AppendBatchRequest;
14use crate::request::AppendExternalRequest;
15use crate::request::AppendRequest;
16use crate::request::CloseStreamRequest;
17use crate::request::CreateStreamExternalRequest;
18use crate::request::CreateStreamRequest;
19use crate::request::DeleteStreamRequest;
20use crate::request::FlushColdRequest;
21use crate::request::PublishSnapshotRequest;
22use crate::request::StreamAppendCount;
23
24#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
25pub struct GroupSnapshot {
26 pub placement: ShardPlacement,
27 pub group_commit_index: u64,
28 pub stream_snapshot: StreamSnapshot,
29 pub stream_append_counts: Vec<StreamAppendCount>,
30}
31
32#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
33#[serde(tag = "command", rename_all = "snake_case")]
34pub enum GroupWriteCommand {
35 CreateStream {
36 stream_id: BucketStreamId,
37 content_type: String,
38 initial_payload: Bytes,
39 close_after: bool,
40 stream_seq: Option<String>,
41 producer: Option<ProducerRequest>,
42 stream_ttl_seconds: Option<u64>,
43 stream_expires_at_ms: Option<u64>,
44 forked_from: Option<BucketStreamId>,
45 fork_offset: Option<u64>,
46 now_ms: u64,
47 },
48 CreateExternal {
49 stream_id: BucketStreamId,
50 content_type: String,
51 initial_payload: ExternalPayloadRef,
52 close_after: bool,
53 stream_seq: Option<String>,
54 producer: Option<ProducerRequest>,
55 stream_ttl_seconds: Option<u64>,
56 stream_expires_at_ms: Option<u64>,
57 forked_from: Option<BucketStreamId>,
58 fork_offset: Option<u64>,
59 now_ms: u64,
60 },
61 Append {
62 stream_id: BucketStreamId,
63 content_type: String,
64 payload: Bytes,
65 close_after: bool,
66 stream_seq: Option<String>,
67 producer: Option<ProducerRequest>,
68 now_ms: u64,
69 },
70 AppendExternal {
71 stream_id: BucketStreamId,
72 content_type: String,
73 payload: ExternalPayloadRef,
74 close_after: bool,
75 stream_seq: Option<String>,
76 producer: Option<ProducerRequest>,
77 now_ms: u64,
78 },
79 AppendBatch {
80 stream_id: BucketStreamId,
81 content_type: String,
82 payloads: Vec<Bytes>,
83 producer: Option<ProducerRequest>,
84 now_ms: u64,
85 },
86 PublishSnapshot {
87 stream_id: BucketStreamId,
88 snapshot_offset: u64,
89 content_type: String,
90 payload: Bytes,
91 now_ms: u64,
92 },
93 TouchStreamAccess {
94 stream_id: BucketStreamId,
95 now_ms: u64,
96 renew_ttl: bool,
97 },
98 AddForkRef {
99 stream_id: BucketStreamId,
100 now_ms: u64,
101 },
102 ReleaseForkRef {
103 stream_id: BucketStreamId,
104 },
105 FlushCold {
106 stream_id: BucketStreamId,
107 chunk: ColdChunkRef,
108 },
109 CloseStream {
110 stream_id: BucketStreamId,
111 stream_seq: Option<String>,
112 producer: Option<ProducerRequest>,
113 now_ms: u64,
114 },
115 DeleteStream {
116 stream_id: BucketStreamId,
117 },
118 AckColdGc {
119 up_to_seq: u64,
120 },
121 Batch {
122 commands: Vec<GroupWriteCommand>,
123 },
124}
125
126impl From<CreateStreamRequest> for GroupWriteCommand {
127 fn from(request: CreateStreamRequest) -> Self {
128 Self::CreateStream {
129 stream_id: request.stream_id,
130 content_type: request.content_type,
131 initial_payload: request.initial_payload,
132 close_after: request.close_after,
133 stream_seq: request.stream_seq,
134 producer: request.producer,
135 stream_ttl_seconds: request.stream_ttl_seconds,
136 stream_expires_at_ms: request.stream_expires_at_ms,
137 forked_from: request.forked_from,
138 fork_offset: request.fork_offset,
139 now_ms: request.now_ms,
140 }
141 }
142}
143
144impl From<&CreateStreamRequest> for GroupWriteCommand {
145 fn from(request: &CreateStreamRequest) -> Self {
146 Self::CreateStream {
147 stream_id: request.stream_id.clone(),
148 content_type: request.content_type.clone(),
149 initial_payload: request.initial_payload.clone(),
150 close_after: request.close_after,
151 stream_seq: request.stream_seq.clone(),
152 producer: request.producer.clone(),
153 stream_ttl_seconds: request.stream_ttl_seconds,
154 stream_expires_at_ms: request.stream_expires_at_ms,
155 forked_from: request.forked_from.clone(),
156 fork_offset: request.fork_offset,
157 now_ms: request.now_ms,
158 }
159 }
160}
161
162impl From<CreateStreamExternalRequest> for GroupWriteCommand {
163 fn from(request: CreateStreamExternalRequest) -> Self {
164 Self::CreateExternal {
165 stream_id: request.stream_id,
166 content_type: request.content_type,
167 initial_payload: request.initial_payload,
168 close_after: request.close_after,
169 stream_seq: request.stream_seq,
170 producer: request.producer,
171 stream_ttl_seconds: request.stream_ttl_seconds,
172 stream_expires_at_ms: request.stream_expires_at_ms,
173 forked_from: request.forked_from,
174 fork_offset: request.fork_offset,
175 now_ms: request.now_ms,
176 }
177 }
178}
179
180impl From<&CreateStreamExternalRequest> for GroupWriteCommand {
181 fn from(request: &CreateStreamExternalRequest) -> Self {
182 Self::CreateExternal {
183 stream_id: request.stream_id.clone(),
184 content_type: request.content_type.clone(),
185 initial_payload: request.initial_payload.clone(),
186 close_after: request.close_after,
187 stream_seq: request.stream_seq.clone(),
188 producer: request.producer.clone(),
189 stream_ttl_seconds: request.stream_ttl_seconds,
190 stream_expires_at_ms: request.stream_expires_at_ms,
191 forked_from: request.forked_from.clone(),
192 fork_offset: request.fork_offset,
193 now_ms: request.now_ms,
194 }
195 }
196}
197
198impl From<AppendRequest> for GroupWriteCommand {
199 fn from(request: AppendRequest) -> Self {
200 Self::Append {
201 stream_id: request.stream_id,
202 content_type: request.content_type,
203 payload: request.payload,
204 close_after: request.close_after,
205 stream_seq: request.stream_seq,
206 producer: request.producer,
207 now_ms: request.now_ms,
208 }
209 }
210}
211
212impl From<&AppendRequest> for GroupWriteCommand {
213 fn from(request: &AppendRequest) -> Self {
214 Self::Append {
215 stream_id: request.stream_id.clone(),
216 content_type: request.content_type.clone(),
217 payload: request.payload.clone(),
218 close_after: request.close_after,
219 stream_seq: request.stream_seq.clone(),
220 producer: request.producer.clone(),
221 now_ms: request.now_ms,
222 }
223 }
224}
225
226impl From<AppendExternalRequest> for GroupWriteCommand {
227 fn from(request: AppendExternalRequest) -> Self {
228 Self::AppendExternal {
229 stream_id: request.stream_id,
230 content_type: request.content_type,
231 payload: request.payload,
232 close_after: request.close_after,
233 stream_seq: request.stream_seq,
234 producer: request.producer,
235 now_ms: request.now_ms,
236 }
237 }
238}
239
240impl From<&AppendExternalRequest> for GroupWriteCommand {
241 fn from(request: &AppendExternalRequest) -> Self {
242 Self::AppendExternal {
243 stream_id: request.stream_id.clone(),
244 content_type: request.content_type.clone(),
245 payload: request.payload.clone(),
246 close_after: request.close_after,
247 stream_seq: request.stream_seq.clone(),
248 producer: request.producer.clone(),
249 now_ms: request.now_ms,
250 }
251 }
252}
253
254impl From<AppendBatchRequest> for GroupWriteCommand {
255 fn from(request: AppendBatchRequest) -> Self {
256 Self::AppendBatch {
257 stream_id: request.stream_id,
258 content_type: request.content_type,
259 payloads: request.payloads,
260 producer: request.producer,
261 now_ms: request.now_ms,
262 }
263 }
264}
265
266impl From<&AppendBatchRequest> for GroupWriteCommand {
267 fn from(request: &AppendBatchRequest) -> Self {
268 Self::AppendBatch {
269 stream_id: request.stream_id.clone(),
270 content_type: request.content_type.clone(),
271 payloads: request.payloads.clone(),
272 producer: request.producer.clone(),
273 now_ms: request.now_ms,
274 }
275 }
276}
277
278impl From<PublishSnapshotRequest> for GroupWriteCommand {
279 fn from(request: PublishSnapshotRequest) -> Self {
280 Self::PublishSnapshot {
281 stream_id: request.stream_id,
282 snapshot_offset: request.snapshot_offset,
283 content_type: request.content_type,
284 payload: request.payload,
285 now_ms: request.now_ms,
286 }
287 }
288}
289
290impl From<&PublishSnapshotRequest> for GroupWriteCommand {
291 fn from(request: &PublishSnapshotRequest) -> Self {
292 Self::PublishSnapshot {
293 stream_id: request.stream_id.clone(),
294 snapshot_offset: request.snapshot_offset,
295 content_type: request.content_type.clone(),
296 payload: request.payload.clone(),
297 now_ms: request.now_ms,
298 }
299 }
300}
301
302impl From<CloseStreamRequest> for GroupWriteCommand {
303 fn from(request: CloseStreamRequest) -> Self {
304 Self::CloseStream {
305 stream_id: request.stream_id,
306 stream_seq: request.stream_seq,
307 producer: request.producer,
308 now_ms: request.now_ms,
309 }
310 }
311}
312
313impl From<&CloseStreamRequest> for GroupWriteCommand {
314 fn from(request: &CloseStreamRequest) -> Self {
315 Self::CloseStream {
316 stream_id: request.stream_id.clone(),
317 stream_seq: request.stream_seq.clone(),
318 producer: request.producer.clone(),
319 now_ms: request.now_ms,
320 }
321 }
322}
323
324impl From<DeleteStreamRequest> for GroupWriteCommand {
325 fn from(request: DeleteStreamRequest) -> Self {
326 Self::DeleteStream {
327 stream_id: request.stream_id,
328 }
329 }
330}
331
332impl From<&DeleteStreamRequest> for GroupWriteCommand {
333 fn from(request: &DeleteStreamRequest) -> Self {
334 Self::DeleteStream {
335 stream_id: request.stream_id.clone(),
336 }
337 }
338}
339
340impl From<FlushColdRequest> for GroupWriteCommand {
341 fn from(request: FlushColdRequest) -> Self {
342 Self::FlushCold {
343 stream_id: request.stream_id,
344 chunk: request.chunk,
345 }
346 }
347}
348
349impl From<&FlushColdRequest> for GroupWriteCommand {
350 fn from(request: &FlushColdRequest) -> Self {
351 Self::FlushCold {
352 stream_id: request.stream_id.clone(),
353 chunk: request.chunk.clone(),
354 }
355 }
356}
357
358impl fmt::Display for GroupWriteCommand {
359 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
360 match self {
361 Self::CreateStream { stream_id, .. } => {
362 write!(f, "create_stream:{stream_id}")
363 }
364 Self::CreateExternal {
365 stream_id,
366 initial_payload,
367 ..
368 } => {
369 write!(
370 f,
371 "create_external:{stream_id}:{} bytes",
372 initial_payload.payload_len
373 )
374 }
375 Self::Append {
376 stream_id, payload, ..
377 } => {
378 write!(f, "append:{stream_id}:{} bytes", payload.len())
379 }
380 Self::AppendExternal {
381 stream_id, payload, ..
382 } => {
383 write!(
384 f,
385 "append_external:{stream_id}:{} bytes",
386 payload.payload_len
387 )
388 }
389 Self::AppendBatch {
390 stream_id,
391 payloads,
392 ..
393 } => {
394 write!(f, "append_batch:{stream_id}:{} items", payloads.len())
395 }
396 Self::PublishSnapshot {
397 stream_id,
398 snapshot_offset,
399 payload,
400 ..
401 } => {
402 write!(
403 f,
404 "publish_snapshot:{stream_id}:{snapshot_offset}:{} bytes",
405 payload.len()
406 )
407 }
408 Self::TouchStreamAccess {
409 stream_id,
410 renew_ttl,
411 ..
412 } => {
413 write!(f, "touch_stream_access:{stream_id}:renew_ttl={renew_ttl}")
414 }
415 Self::AddForkRef { stream_id, .. } => {
416 write!(f, "add_fork_ref:{stream_id}")
417 }
418 Self::ReleaseForkRef { stream_id } => {
419 write!(f, "release_fork_ref:{stream_id}")
420 }
421 Self::FlushCold { stream_id, chunk } => {
422 write!(
423 f,
424 "flush_cold:{stream_id}:{}..{}",
425 chunk.start_offset, chunk.end_offset
426 )
427 }
428 Self::CloseStream { stream_id, .. } => {
429 write!(f, "close_stream:{stream_id}")
430 }
431 Self::DeleteStream { stream_id } => {
432 write!(f, "delete_stream:{stream_id}")
433 }
434 Self::AckColdGc { up_to_seq } => {
435 write!(f, "ack_cold_gc:up_to_seq={up_to_seq}")
436 }
437 Self::Batch { commands } => {
438 write!(f, "batch:{} commands", commands.len())
439 }
440 }
441 }
442}