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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
// Copyright (c) 2021 Anatoly Ikorsky
//
// Licensed under the Apache License, Version 2.0
// <LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0> or the MIT
// license <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
// option. All files in the project carrying such notice may not be copied,
// modified, or distributed except according to those terms.

use std::{cmp::min, io};

use saturating::Saturating as S;

use crate::{
    binlog::{
        consts::{BinlogVersion, EventType, Gno, GtidFlags},
        BinlogCtx, BinlogEvent, BinlogStruct,
    },
    io::ParseBuf,
    misc::raw::{int::*, RawConst, RawFlags},
    proto::{MyDeserialize, MySerialize},
};

use super::BinlogEventHeader;

define_const!(
    ConstU8,
    LogicalTimestampTypecode,
    InvalidLogicalTimestampTypecode("Invalid logical timestamp typecode value for GTID event"),
    2
);

/// GTID stands for Global Transaction IDentifier.
#[derive(Debug, Clone, Eq, PartialEq, Hash)]
pub struct GtidEvent {
    /// Raw flags value.
    flags: RawFlags<GtidFlags, u8>,
    /// UUID representing the SID.
    sid: [u8; Self::ENCODED_SID_LENGTH],
    /// Group number, second component of GTID.
    ///
    ///
    /// Should be an integer between `MIN_GNO` and `MAX_GNO` for GtidEvent
    /// or `0` for AnonymousGtidEvent.
    gno: RawConst<LeU64, Gno>,
    /// If defined, then always equal to the constant [`GtidEvent::LOGICAL_TIMESTAMP_TYPECODE`].
    ///
    /// May be missing for 5.6. Will have different value on 5.7.4 and earlier (ignored).
    lc_typecode: Option<LogicalTimestampTypecode>,
    /// Store the transaction's commit parent `sequence_number`.
    last_committed: RawInt<LeU64>,
    /// The transaction's logical timestamp assigned at prepare phase.
    ///
    /// If it isn't `0` then it must be greater than `last_committed` timestamp.
    sequence_number: RawInt<LeU64>,
    /// Timestamp when the transaction was committed on the nearest master.
    immediate_commit_timestamp: RawInt<LeU56>,
    /// Timestamp when the transaction was committed on the originating master.
    original_commit_timestamp: RawInt<LeU56>,
    /// The packed transaction's length in bytes, including the Gtid.
    tx_length: RawInt<LenEnc>,
    /// The version of the server where the transaction was originally executed.
    original_server_version: RawInt<LeU32>,
    /// The version of the immediate server.
    immediate_server_version: RawInt<LeU32>,
}

impl GtidEvent {
    pub const POST_HEADER_LENGTH: usize = 1 + Self::ENCODED_SID_LENGTH + 8 + 1 + 16;
    pub const ENCODED_SID_LENGTH: usize = 16;
    pub const LOGICAL_TIMESTAMP_TYPECODE: u8 = 2;
    pub const IMMEDIATE_COMMIT_TIMESTAMP_LENGTH: usize = 7;
    pub const ORIGINAL_COMMIT_TIMESTAMP_LENGTH: usize = 7;
    pub const UNDEFINED_SERVER_VERSION: u32 = 999_999;
    pub const IMMEDIATE_SERVER_VERSION_LENGTH: usize = 4;

    pub fn new(sid: [u8; Self::ENCODED_SID_LENGTH], gno: u64) -> Self {
        Self {
            flags: Default::default(),
            sid,
            gno: RawConst::new(gno),
            lc_typecode: Some(LogicalTimestampTypecode::default()),
            last_committed: Default::default(),
            sequence_number: Default::default(),
            immediate_commit_timestamp: Default::default(),
            original_commit_timestamp: Default::default(),
            tx_length: Default::default(),
            original_server_version: Default::default(),
            immediate_server_version: Default::default(),
        }
    }

    /// Defines the `flags` value.
    pub fn with_flags(mut self, flags: GtidFlags) -> Self {
        self.flags = RawFlags::new(flags.bits());
        self
    }

    /// Returns the raw `flags` value.
    pub fn flags_raw(&self) -> u8 {
        self.flags.0
    }

    /// Returns the `flags` value. Unknown bits will be truncated.
    ///
    /// `00000001` – Transaction may have changes logged with SBR.
    ///
    /// In 5.6, 5.7.0-5.7.18, and 8.0.0-8.0.1, this flag is always set. Starting in 5.7.19 and
    /// 8.0.2, this flag is cleared if the transaction only contains row events.
    /// It is set if any part of the transaction is written in statement format.
    pub fn flags(&self) -> GtidFlags {
        self.flags.get()
    }

    /// Defines the `sid` value.
    pub fn with_sid(mut self, sid: [u8; Self::ENCODED_SID_LENGTH]) -> Self {
        self.sid = sid;
        self
    }

    /// Returns the `sid` value.
    ///
    /// `sid` is the UUID representing the SID.
    pub fn sid(&self) -> [u8; Self::ENCODED_SID_LENGTH] {
        self.sid
    }

    /// Defines the `gno` value.
    pub fn with_gno(mut self, gno: u64) -> Self {
        self.gno = RawConst::new(gno);
        self
    }

    /// Returns the `gno` value.
    ///
    /// `gno` is a group number, second component of GTID.
    pub fn gno(&self) -> u64 {
        self.gno.0
    }

    /// Returns the `lc_typecode` value.
    ///
    /// `lc_typecode` is the type of logical timestamp used in the logical clock fields.
    pub fn lc_typecode(&self) -> Option<u8> {
        self.lc_typecode.as_ref().map(|x| x.value())
    }

    /// Sets the `lc_typecode` value to [`GtidEvent::LOGICAL_TIMESTAMP_TYPECODE`].
    ///
    /// This is already by default, but `lc_typecode` might be `None` if `Self` is obtained
    /// from an old MySql server via [`MyDeserialize::deserialize`].
    pub fn with_lc_typecode(mut self) -> Self {
        self.lc_typecode = Some(LogicalTimestampTypecode::default());
        self
    }

    /// Sets the `last_committed` value.
    pub fn with_last_committed(mut self, last_committed: u64) -> Self {
        self.last_committed = RawInt::new(last_committed);
        self
    }

    /// Returns the `last_committed` value.
    ///
    /// `last_committed` stores the transaction's commit parent `sequence_number`.
    pub fn last_committed(&self) -> u64 {
        self.last_committed.0
    }

    /// Sets the `sequence_number` value.
    pub fn with_sequence_number(mut self, sequence_number: u64) -> Self {
        self.sequence_number = RawInt::new(sequence_number);
        self
    }

    /// Returns the `sequence_number` value.
    ///
    /// `sequence_number` is the transaction's logical timestamp assigned at prepare phase.
    pub fn sequence_number(&self) -> u64 {
        self.sequence_number.0
    }

    /// Sets the `immediate_commit_timestamp` value.
    pub fn with_immediate_commit_timestamp(mut self, immediate_commit_timestamp: u64) -> Self {
        self.immediate_commit_timestamp = RawInt::new(immediate_commit_timestamp);
        self
    }

    /// Returns the `immediate_commit_timestamp` value.
    ///
    /// `immediate_commit_timestamp` is a timestamp of commit on the immediate master.
    pub fn immediate_commit_timestamp(&self) -> u64 {
        self.immediate_commit_timestamp.0
    }

    /// Sets the `original_commit_timestamp` value.
    pub fn with_original_commit_timestamp(mut self, original_commit_timestamp: u64) -> Self {
        self.original_commit_timestamp = RawInt::new(original_commit_timestamp);
        self
    }

    /// Returns the `original_commit_timestamp` value.
    ///
    /// `original_commit_timestamp` is the timestamp of commit on the originating master.
    pub fn original_commit_timestamp(&self) -> u64 {
        self.original_commit_timestamp.0
    }

    /// Sets the `tx_length` value.
    pub fn with_tx_length(mut self, tx_length: u64) -> Self {
        self.tx_length = RawInt::new(tx_length);
        self
    }

    /// Returns the `tx_length` value.
    ///
    /// `tx_length` is the packed transaction's length in bytes, including the Gtid.
    pub fn tx_length(&self) -> u64 {
        self.tx_length.0
    }

    /// Sets the `original_server_version` value.
    pub fn with_original_server_version(mut self, original_server_version: u32) -> Self {
        self.original_server_version = RawInt::new(original_server_version);
        self
    }

    /// Returns the `original_server_version` value.
    ///
    /// `original_server_version` is the version of the server where the transaction was originally
    /// executed.
    pub fn original_server_version(&self) -> u32 {
        self.original_server_version.0
    }

    /// Sets the `immediate_server_version` value.
    pub fn with_immediate_server_version(mut self, immediate_server_version: u32) -> Self {
        self.immediate_server_version = RawInt::new(immediate_server_version);
        self
    }

    /// Returns the `immediate_server_version` value.
    ///
    /// `immediate_server_version` is the server version of the immediate server.
    pub fn immediate_server_version(&self) -> u32 {
        self.immediate_server_version.0
    }
}

impl<'de> MyDeserialize<'de> for GtidEvent {
    const SIZE: Option<usize> = None;
    type Ctx = BinlogCtx<'de>;

    fn deserialize(_ctx: Self::Ctx, buf: &mut ParseBuf<'de>) -> io::Result<Self> {
        let mut sbuf: ParseBuf = buf.parse(1 + Self::ENCODED_SID_LENGTH + 8)?;
        let flags = sbuf.parse_unchecked(())?;
        let sid: [u8; Self::ENCODED_SID_LENGTH] = sbuf.parse_unchecked(())?;
        let gno = sbuf.parse_unchecked(())?;

        let mut lc_typecode = None;
        let mut last_committed = RawInt::new(0);
        let mut sequence_number = RawInt::new(0);
        let mut immediate_commit_timestamp = RawInt::new(0);
        let mut original_commit_timestamp = RawInt::new(0);
        let mut tx_length = RawInt::new(0);

        let mut original_server_version = RawInt::new(Self::UNDEFINED_SERVER_VERSION);
        let mut immediate_server_version = RawInt::new(Self::UNDEFINED_SERVER_VERSION);

        // Buf will be empty for MySql 5.6. Condition will be false for MySql <= 5.7.4
        if !buf.is_empty() && buf.0[0] == Self::LOGICAL_TIMESTAMP_TYPECODE {
            lc_typecode = Some(buf.parse_unchecked(())?);

            let mut sbuf: ParseBuf = buf.parse(16)?;
            last_committed = sbuf.parse_unchecked(())?;
            sequence_number = sbuf.parse_unchecked(())?;

            if buf.len() >= Self::IMMEDIATE_COMMIT_TIMESTAMP_LENGTH {
                immediate_commit_timestamp = buf.parse_unchecked(())?;
                if immediate_commit_timestamp.0 & (1 << 55) != 0 {
                    immediate_commit_timestamp.0 &= !(1 << 55);
                    original_commit_timestamp = buf.parse(())?;
                } else {
                    // The transaction originated in the previous server
                    original_commit_timestamp = immediate_commit_timestamp;
                }
            }

            if !buf.is_empty() {
                tx_length = buf.parse_unchecked(())?;
            }

            if buf.len() >= Self::IMMEDIATE_SERVER_VERSION_LENGTH {
                immediate_server_version = buf.parse_unchecked(())?;
                if immediate_server_version.0 & (1 << 31) != 0 {
                    immediate_server_version.0 &= !(1 << 31);
                    original_server_version = buf.parse(())?;
                } else {
                    original_server_version = immediate_server_version;
                }
            }
        }

        Ok(Self {
            flags,
            sid,
            gno,
            lc_typecode,
            last_committed,
            sequence_number,
            immediate_commit_timestamp,
            original_commit_timestamp,
            tx_length,
            original_server_version,
            immediate_server_version,
        })
    }
}

impl<'de> MySerialize for GtidEvent {
    fn serialize(&self, buf: &mut Vec<u8>) {
        self.flags.serialize(&mut *buf);
        self.sid.serialize(&mut *buf);
        self.gno.serialize(&mut *buf);
        match self.lc_typecode {
            Some(lc_typecode) => lc_typecode.serialize(&mut *buf),
            None => return,
        };
        self.last_committed.serialize(&mut *buf);
        self.sequence_number.serialize(&mut *buf);

        let mut immediate_commit_timestamp_with_flag = *self.immediate_commit_timestamp;
        if self.immediate_commit_timestamp != self.original_commit_timestamp {
            immediate_commit_timestamp_with_flag |= 1 << 55;
        } else {
            immediate_commit_timestamp_with_flag &= !(1 << 55);
        }
        RawInt::<LeU56>::new(immediate_commit_timestamp_with_flag).serialize(&mut *buf);

        if self.immediate_commit_timestamp != self.original_commit_timestamp {
            self.original_commit_timestamp.serialize(&mut *buf);
        }

        self.tx_length.serialize(&mut *buf);

        let mut immediate_server_version_with_flag = *self.immediate_server_version;
        if self.immediate_server_version != self.original_server_version {
            immediate_server_version_with_flag |= 1 << 31;
        } else {
            immediate_server_version_with_flag &= !(1 << 31);
        }
        RawInt::<LeU32>::new(immediate_server_version_with_flag).serialize(&mut *buf);

        if self.immediate_server_version != self.original_server_version {
            self.original_server_version.serialize(&mut *buf);
        }
    }
}

impl<'a> BinlogStruct<'a> for GtidEvent {
    fn len(&self, _version: BinlogVersion) -> usize {
        let mut len = S(0);

        // post header
        len += S(1); // flags
        len += S(Self::ENCODED_SID_LENGTH); // sid
        len += S(8); // gno
        len += S(1); // lc_typecode
        len += S(8); // last_committed
        len += S(8); // sequence_number

        len += S(7); // immediate_commit_timestamp
        if self.immediate_commit_timestamp != self.original_commit_timestamp {
            len += S(7); // original_commit_timestamp
        }

        len += S(crate::misc::lenenc_int_len(*self.tx_length as u64) as usize); // tx_length
        len += S(4); // immediate_server_version
        if self.immediate_server_version != self.original_server_version {
            len += S(4); // original_server_version
        }

        min(len.0, u32::MAX as usize - BinlogEventHeader::LEN)
    }
}

impl<'a> BinlogEvent<'a> for GtidEvent {
    const EVENT_TYPE: EventType = EventType::GTID_EVENT;
}