triplox-client 0.1.0-alpha.5

HTTP/2 client for Triplox, a Datalog database.
Documentation 
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

//! Shared protocol constants and types used by the storage codec, the
//! HTTP/2 server, and the msgpack wire codec.
//!
//! The actual wire encoding lives in [`crate::msgpack_codec`]; this module
//! holds the data-type tag taxonomy (also used by the storage codec to keep
//! a single set of type identifiers across storage and wire), the error code
//! enum, and the column description used in query response schemas.

use anyhow::{anyhow, bail, Result};
use chrono::{DateTime, TimeZone, Utc};

use crate::ops::DataType;

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

/// Convert microseconds since Unix epoch to a `DateTime<Utc>`.
///
/// Uses `div_euclid`/`rem_euclid` so that pre-epoch (negative) timestamps
/// are decoded correctly — Rust's `/` truncates toward zero which gives
/// wrong results for negative values.
pub fn micros_to_datetime(micros: i64) -> Result<DateTime<Utc>> {
    let secs = micros.div_euclid(1_000_000);
    let nanos = (micros.rem_euclid(1_000_000) as u32) * 1000;
    Utc.timestamp_opt(secs, nanos)
        .single()
        .ok_or_else(|| anyhow!("Invalid timestamp: {} micros", micros))
}

// ---------------------------------------------------------------------------
// Constants
// ---------------------------------------------------------------------------

/// Default maximum HTTP body size (64 MB).
pub const DEFAULT_MAX_MESSAGE_SIZE: u32 = 64 * 1024 * 1024;

/// `ErrorResponse` severity byte for non-fatal errors.
pub const SEVERITY_ERROR: u8 = b'E';
/// `ErrorResponse` severity byte for fatal errors.
pub const SEVERITY_FATAL: u8 = b'F';

// DataType tag bytes — shared between the storage codec (`crate::codec`)
// and column descriptions in query responses.
pub const TAG_BIG_INT: u8 = 1;
pub const TAG_BOOLEAN: u8 = 2;
pub const TAG_BYTES: u8 = 3;
pub const TAG_DOUBLE: u8 = 4;
pub const TAG_FLOAT: u8 = 5;
pub const TAG_INSTANT: u8 = 6;
pub const TAG_LONG: u8 = 7;
/// Reserved for a future `DataType::Ref` variant.
pub const TAG_REF: u8 = 8;
pub const TAG_STRING: u8 = 9;
pub const TAG_UUID: u8 = 10;
pub const TAG_VECTOR: u8 = 11;
pub const TAG_MAP: u8 = 12;
pub const TAG_KEYWORD: u8 = 13;
/// Column-description-only tag for unions of concrete types.
pub const TAG_UNION: u8 = 127;
/// Column-description-only tag for indeterminate column types.
pub const TAG_UNKNOWN: u8 = 255;

// ---------------------------------------------------------------------------
// Error Codes
// ---------------------------------------------------------------------------

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u16)]
pub enum ErrorCode {
    // Connection errors (1xxx)
    ProtocolVersionMismatch = 1000,
    InvalidStartup = 1001,
    // Query errors (2xxx)
    ParseError = 2000,
    QueryError = 2001,
    InvalidQuery = 2002,
    EmptyQuery = 2003,
    // Transaction errors (3xxx)
    TxError = 3000,
    TxAborted = 3001,
    TxNotIndexed = 3002,
    // Internal/protocol errors (4xxx)
    InternalError = 4000,
    MessageTooLarge = 4001,
    InvalidMessageType = 4002,
    QueryCancelled = 4003,
    ServerShuttingDown = 4004,
}

impl ErrorCode {
    pub fn from_u16(code: u16) -> Result<Self> {
        match code {
            1000 => Ok(ErrorCode::ProtocolVersionMismatch),
            1001 => Ok(ErrorCode::InvalidStartup),
            2000 => Ok(ErrorCode::ParseError),
            2001 => Ok(ErrorCode::QueryError),
            2002 => Ok(ErrorCode::InvalidQuery),
            2003 => Ok(ErrorCode::EmptyQuery),
            3000 => Ok(ErrorCode::TxError),
            3001 => Ok(ErrorCode::TxAborted),
            3002 => Ok(ErrorCode::TxNotIndexed),
            4000 => Ok(ErrorCode::InternalError),
            4001 => Ok(ErrorCode::MessageTooLarge),
            4002 => Ok(ErrorCode::InvalidMessageType),
            4003 => Ok(ErrorCode::QueryCancelled),
            4004 => Ok(ErrorCode::ServerShuttingDown),
            _ => bail!("Unknown error code: {}", code),
        }
    }

    pub fn as_u16(self) -> u16 {
        self as u16
    }
}

// ---------------------------------------------------------------------------
// Column Description
// ---------------------------------------------------------------------------

#[derive(Debug, Clone, PartialEq)]
pub struct ColumnDescription {
    pub name: String,
    pub data_type: u8,
    /// For Union columns (`data_type == TAG_UNION`), the concrete member type
    /// tags. `None` for concrete types and `TAG_UNKNOWN`.
    pub members: Option<Vec<u8>>,
}

// ---------------------------------------------------------------------------
// DataType tag dispatch
// ---------------------------------------------------------------------------

/// Returns the DataType tag byte for a value (see `TAG_*` constants).
pub fn data_type_tag(dt: &DataType) -> u8 {
    match dt {
        DataType::BigInt(_) => TAG_BIG_INT,
        DataType::Boolean(_) => TAG_BOOLEAN,
        DataType::Bytes(_) => TAG_BYTES,
        DataType::Double(_) => TAG_DOUBLE,
        DataType::Float(_) => TAG_FLOAT,
        DataType::Instant(_) => TAG_INSTANT,
        DataType::Keyword(_) => TAG_KEYWORD,
        DataType::Long(_) => TAG_LONG,
        DataType::String(_) => TAG_STRING,
        DataType::Uuid(_) => TAG_UUID,
        DataType::Vector(_) => TAG_VECTOR,
        DataType::Map(_) => TAG_MAP,
    }
}