reddb_rql/parser/
queue.rs

1//! Parser for QUEUE commands and CREATE/DROP QUEUE
2
3use super::error::ParseError;
4use super::Parser;
5use crate::ast::{
6    AlterQueueQuery, CreateQueueQuery, DropQueueQuery, QueryExpr, QueueAvailability, QueueCommand,
7    QueueMode, QueueSide, DEFAULT_QUEUE_IN_FLIGHT_CAP_PER_GROUP, DEFAULT_QUEUE_LOCK_DEADLINE_MS,
8    DEFAULT_QUEUE_MAX_ATTEMPTS,
9};
10use crate::lexer::Token;
11
12impl<'a> Parser<'a> {
13    /// Parse CREATE QUEUE body (after CREATE QUEUE consumed)
14    pub fn parse_create_queue_body(&mut self) -> Result<QueryExpr, ParseError> {
15        let if_not_exists = self.match_if_not_exists()?;
16        let name = self.expect_ident()?;
17
18        let mut mode = QueueMode::Work;
19        let mut priority = false;
20        let mut max_size = None;
21        let mut ttl_ms = None;
22        let mut dlq = None;
23        let mut max_attempts = DEFAULT_QUEUE_MAX_ATTEMPTS;
24        let mut lock_deadline_ms = DEFAULT_QUEUE_LOCK_DEADLINE_MS;
25        let mut in_flight_cap_per_group = DEFAULT_QUEUE_IN_FLIGHT_CAP_PER_GROUP;
26        let mut retry_delay_ms: Option<u64> = None;
27        let mut dedup_window_ms: Option<u64> = None;
28
29        // Parse optional clauses in any order
30        loop {
31            if let Some(parsed_mode) = self.consume_queue_mode()? {
32                mode = parsed_mode;
33            } else if self.consume(&Token::Priority)? {
34                priority = true;
35            } else if self.consume_ident_ci("MAX_SIZE")? || self.consume_ident_ci("MAXSIZE")? {
36                max_size = Some(self.parse_positive_integer("MAX_SIZE")? as usize);
37            } else if self.consume_ident_ci("MAX_ATTEMPTS")?
38                || self.consume_ident_ci("MAXATTEMPTS")?
39            {
40                max_attempts = self.parse_integer()?.max(1) as u32;
41            } else if self.consume_ident_ci("LOCK_DEADLINE_MS")? {
42                lock_deadline_ms = self.parse_integer()?.max(1) as u64;
43            } else if self.consume_ident_ci("IN_FLIGHT_CAP_PER_GROUP")? {
44                in_flight_cap_per_group = self.parse_integer()?.max(1) as u32;
45            } else if self.consume_ident_ci("RETRY_DELAY")? {
46                let value = self.parse_float()?;
47                let unit = self.parse_queue_duration_unit()?;
48                retry_delay_ms = Some((value * unit).max(0.0) as u64);
49            } else if self.consume_ident_ci("DEDUP_WINDOW")? {
50                let value = self.parse_float()?;
51                let unit = self.parse_queue_duration_unit()?;
52                dedup_window_ms = Some((value * unit).max(0.0) as u64);
53            } else if self.consume(&Token::With)? {
54                if self.consume_ident_ci("EVENTS")? {
55                    return Err(ParseError::new(
56                        "queues cannot have event subscriptions".to_string(),
57                        self.position(),
58                    ));
59                } else if self.consume_ident_ci("TTL")? {
60                    let value = self.parse_float()?;
61                    let unit = self.parse_queue_duration_unit()?;
62                    ttl_ms = Some((value * unit) as u64);
63                } else if self.consume_ident_ci("DLQ")? {
64                    dlq = Some(self.expect_ident()?);
65                }
66            } else {
67                break;
68            }
69        }
70
71        Ok(QueryExpr::CreateQueue(CreateQueueQuery {
72            name,
73            mode,
74            priority,
75            max_size,
76            ttl_ms,
77            dlq,
78            max_attempts,
79            lock_deadline_ms,
80            in_flight_cap_per_group,
81            if_not_exists,
82            retry_delay_ms,
83            dedup_window_ms,
84        }))
85    }
86
87    /// Parse ALTER QUEUE body (after ALTER QUEUE consumed)
88    ///
89    /// Syntax: `ALTER QUEUE <name> SET <clause>` where `<clause>` is one of
90    ///   `MODE <WORK|STANDARD|FIFO|FANOUT>`
91    ///   `MAX_ATTEMPTS <int>`
92    ///   `LOCK_DEADLINE_MS <int>`
93    ///   `IN_FLIGHT_CAP_PER_GROUP <int>`
94    ///   `DLQ <ident>`
95    pub fn parse_alter_queue_body(&mut self) -> Result<QueryExpr, ParseError> {
96        let name = self.expect_ident()?;
97        if !self.consume(&Token::Set)? && !self.consume_ident_ci("SET")? {
98            return Err(ParseError::expected(
99                vec!["SET"],
100                self.peek(),
101                self.position(),
102            ));
103        }
104
105        let mut alter = AlterQueueQuery {
106            name,
107            ..Default::default()
108        };
109
110        if self.consume(&Token::Mode)? || self.consume_ident_ci("MODE")? {
111            alter.mode = Some(self.parse_queue_mode()?);
112        } else if self.consume_ident_ci("MAX_ATTEMPTS")? || self.consume_ident_ci("MAXATTEMPTS")? {
113            alter.max_attempts = Some(self.parse_integer()?.max(1) as u32);
114        } else if self.consume_ident_ci("LOCK_DEADLINE_MS")? {
115            alter.lock_deadline_ms = Some(self.parse_integer()?.max(1) as u64);
116        } else if self.consume_ident_ci("IN_FLIGHT_CAP_PER_GROUP")? {
117            alter.in_flight_cap_per_group = Some(self.parse_integer()?.max(1) as u32);
118        } else if self.consume_ident_ci("DLQ")? {
119            alter.dlq = Some(self.expect_ident()?);
120        } else if self.consume_ident_ci("RETRY_DELAY")? {
121            let value = self.parse_float()?;
122            let unit = self.parse_queue_duration_unit()?;
123            alter.retry_delay_ms = Some((value * unit).max(0.0) as u64);
124        } else if self.consume_ident_ci("DEDUP_WINDOW")? {
125            let value = self.parse_float()?;
126            let unit = self.parse_queue_duration_unit()?;
127            alter.dedup_window_ms = Some((value * unit).max(0.0) as u64);
128        } else {
129            return Err(ParseError::expected(
130                vec![
131                    "MODE",
132                    "MAX_ATTEMPTS",
133                    "LOCK_DEADLINE_MS",
134                    "IN_FLIGHT_CAP_PER_GROUP",
135                    "DLQ",
136                    "RETRY_DELAY",
137                    "DEDUP_WINDOW",
138                ],
139                self.peek(),
140                self.position(),
141            ));
142        }
143
144        Ok(QueryExpr::AlterQueue(alter))
145    }
146
147    /// Parse DROP QUEUE body (after DROP QUEUE consumed)
148    pub fn parse_drop_queue_body(&mut self) -> Result<QueryExpr, ParseError> {
149        let if_exists = self.match_if_exists()?;
150        let name = self.parse_drop_collection_name()?;
151        Ok(QueryExpr::DropQueue(DropQueueQuery { name, if_exists }))
152    }
153
154    /// Parse QUEUE subcommand (after QUEUE token consumed)
155    pub fn parse_queue_command(&mut self) -> Result<QueryExpr, ParseError> {
156        self.expect(Token::Queue)?;
157
158        match self.peek().clone() {
159            Token::Push => {
160                self.advance()?;
161                let queue = self.expect_ident()?;
162                let value = self.parse_value()?;
163                let (priority, key, available, dedup_key) = self.parse_push_tail_clauses()?;
164                Ok(QueryExpr::QueueCommand(QueueCommand::Push {
165                    queue,
166                    value,
167                    side: QueueSide::Right,
168                    priority,
169                    key,
170                    available,
171                    dedup_key,
172                }))
173            }
174            Token::Pop => {
175                self.advance()?;
176                let queue = self.expect_ident()?;
177                let count = if self.consume(&Token::Count)? {
178                    self.parse_integer()? as usize
179                } else {
180                    1
181                };
182                self.reject_wait_clause("POP")?;
183                Ok(QueryExpr::QueueCommand(QueueCommand::Pop {
184                    queue,
185                    side: QueueSide::Left,
186                    count,
187                }))
188            }
189            Token::Peek => {
190                self.advance()?;
191                let queue = self.expect_ident()?;
192                // Accept either `PEEK <q> COUNT <n>` or a bare `PEEK <q> <n>`.
193                // `consume` has a side effect (advances past COUNT), so the
194                // short-circuit order matters: only peek for a bare integer
195                // when the COUNT keyword was absent.
196                let count =
197                    if self.consume(&Token::Count)? || matches!(self.peek(), Token::Integer(_)) {
198                        self.parse_integer()? as usize
199                    } else {
200                        1
201                    };
202                self.reject_wait_clause("PEEK")?;
203                Ok(QueryExpr::QueueCommand(QueueCommand::Peek { queue, count }))
204            }
205            Token::Ident(ref name) if name.eq_ignore_ascii_case("LEN") => {
206                self.advance()?;
207                let queue = self.expect_ident()?;
208                Ok(QueryExpr::QueueCommand(QueueCommand::Len { queue }))
209            }
210            Token::Ident(ref name) if name.eq_ignore_ascii_case("MOVE") => {
211                self.advance()?;
212                self.expect(Token::From)?;
213                let source = self.expect_ident()?;
214                self.expect(Token::To)?;
215                let destination = self.expect_ident()?;
216                let filter = if self.consume(&Token::Where)? {
217                    Some(self.parse_filter()?)
218                } else {
219                    None
220                };
221                let limit = if self.consume(&Token::Limit)? {
222                    self.parse_positive_integer("LIMIT")? as usize
223                } else if filter.is_some() {
224                    return Err(ParseError::expected(
225                        vec!["LIMIT"],
226                        self.peek(),
227                        self.position(),
228                    ));
229                } else {
230                    1
231                };
232                Ok(QueryExpr::QueueCommand(QueueCommand::Move {
233                    source,
234                    destination,
235                    filter,
236                    limit,
237                }))
238            }
239            Token::Purge => {
240                self.advance()?;
241                let queue = self.expect_ident()?;
242                Ok(QueryExpr::QueueCommand(QueueCommand::Purge { queue }))
243            }
244            Token::Ident(ref name) if name.eq_ignore_ascii_case("LPOP") => {
245                self.advance()?;
246                let queue = self.expect_ident()?;
247                Ok(QueryExpr::QueueCommand(QueueCommand::Pop {
248                    queue,
249                    side: QueueSide::Left,
250                    count: 1,
251                }))
252            }
253            Token::Ident(ref name) if name.eq_ignore_ascii_case("RPOP") => {
254                self.advance()?;
255                let queue = self.expect_ident()?;
256                Ok(QueryExpr::QueueCommand(QueueCommand::Pop {
257                    queue,
258                    side: QueueSide::Right,
259                    count: 1,
260                }))
261            }
262            Token::Ident(ref name) if name.eq_ignore_ascii_case("LPUSH") => {
263                self.advance()?;
264                let queue = self.expect_ident()?;
265                let value = self.parse_value()?;
266                let (_priority, key, available, dedup_key) = self.parse_push_tail_clauses()?;
267                Ok(QueryExpr::QueueCommand(QueueCommand::Push {
268                    queue,
269                    value,
270                    side: QueueSide::Left,
271                    priority: None,
272                    key,
273                    available,
274                    dedup_key,
275                }))
276            }
277            Token::Ident(ref name) if name.eq_ignore_ascii_case("RPUSH") => {
278                self.advance()?;
279                let queue = self.expect_ident()?;
280                let value = self.parse_value()?;
281                let (priority, key, available, dedup_key) = self.parse_push_tail_clauses()?;
282                Ok(QueryExpr::QueueCommand(QueueCommand::Push {
283                    queue,
284                    value,
285                    side: QueueSide::Right,
286                    priority,
287                    key,
288                    available,
289                    dedup_key,
290                }))
291            }
292            Token::Group => {
293                self.advance()?;
294                self.expect(Token::Create)?;
295                let queue = self.expect_ident()?;
296                let group = self.expect_ident()?;
297                Ok(QueryExpr::QueueCommand(QueueCommand::GroupCreate {
298                    queue,
299                    group,
300                }))
301            }
302            Token::Ident(ref name) if name.eq_ignore_ascii_case("READ") => {
303                self.advance()?;
304                let queue = self.expect_ident()?;
305                let group = if self.consume(&Token::Group)? {
306                    Some(self.expect_ident()?)
307                } else {
308                    None
309                };
310                // CONSUMER consumer_name
311                if !self.consume_ident_ci("CONSUMER")? {
312                    return Err(ParseError::expected(
313                        vec!["CONSUMER"],
314                        self.peek(),
315                        self.position(),
316                    ));
317                }
318                let consumer = self.expect_ident()?;
319                let count = if self.consume(&Token::Count)? {
320                    self.parse_integer()? as usize
321                } else {
322                    1
323                };
324                let wait_ms = self.parse_optional_wait_clause()?;
325                Ok(QueryExpr::QueueCommand(QueueCommand::GroupRead {
326                    queue,
327                    group,
328                    consumer,
329                    count,
330                    wait_ms,
331                }))
332            }
333            Token::Ident(ref name) if name.eq_ignore_ascii_case("PENDING") => {
334                self.advance()?;
335                let queue = self.expect_ident()?;
336                self.expect(Token::Group)?;
337                let group = self.expect_ident()?;
338                Ok(QueryExpr::QueueCommand(QueueCommand::Pending {
339                    queue,
340                    group,
341                }))
342            }
343            Token::Ident(ref name) if name.eq_ignore_ascii_case("CLAIM") => {
344                self.advance()?;
345                let queue = self.expect_ident()?;
346                self.expect(Token::Group)?;
347                let group = self.expect_ident()?;
348                if !self.consume_ident_ci("CONSUMER")? {
349                    return Err(ParseError::expected(
350                        vec!["CONSUMER"],
351                        self.peek(),
352                        self.position(),
353                    ));
354                }
355                let consumer = self.expect_ident()?;
356                if !self.consume_ident_ci("MIN_IDLE")? {
357                    return Err(ParseError::expected(
358                        vec!["MIN_IDLE"],
359                        self.peek(),
360                        self.position(),
361                    ));
362                }
363                let min_idle_ms = self.parse_integer()?.max(0) as u64;
364                Ok(QueryExpr::QueueCommand(QueueCommand::Claim {
365                    queue,
366                    group,
367                    consumer,
368                    min_idle_ms,
369                }))
370            }
371            Token::Ack => {
372                self.advance()?;
373                let queue = self.expect_ident()?;
374                let (group, message_id, delivery_id, _delay_ms) =
375                    self.parse_ack_nack_handle(false)?;
376                Ok(QueryExpr::QueueCommand(QueueCommand::Ack {
377                    queue,
378                    group,
379                    message_id,
380                    delivery_id,
381                }))
382            }
383            Token::Nack => {
384                self.advance()?;
385                let queue = self.expect_ident()?;
386                let (group, message_id, delivery_id, delay_ms) =
387                    self.parse_ack_nack_handle(true)?;
388                Ok(QueryExpr::QueueCommand(QueueCommand::Nack {
389                    queue,
390                    group,
391                    message_id,
392                    delivery_id,
393                    delay_ms,
394                }))
395            }
396            _ => Err(ParseError::expected(
397                vec![
398                    "PUSH", "POP", "PEEK", "LEN", "PURGE", "GROUP", "READ", "ACK", "NACK", "LPUSH",
399                    "RPUSH", "LPOP", "RPOP", "PENDING", "CLAIM", "MOVE",
400                ],
401                self.peek(),
402                self.position(),
403            )),
404        }
405    }
406
407    /// Parse the tuple/`delivery_id` handle that follows `ACK <queue>` /
408    /// `NACK <queue>`. Returns `(group, message_id, delivery_id)`:
409    ///
410    /// - `GROUP <group> '<message_id>'`               → legacy tuple, no delivery_id
411    /// - `GROUP <group> '<message_id>' WITH delivery_id = '<base32>'` → both, delivery_id wins (ADR 0026)
412    /// - `WITH delivery_id = '<base32>'`              → delivery_id only; group / message_id empty
413    ///
414    /// Refusing both is a parse error — at least one handle is required.
415    fn parse_ack_nack_handle(
416        &mut self,
417        allow_delay: bool,
418    ) -> Result<(String, String, Option<String>, Option<u64>), ParseError> {
419        let (group, message_id) = if matches!(self.peek(), Token::Group) {
420            self.advance()?;
421            let group = self.expect_ident()?;
422            let message_id = self.parse_string()?;
423            (group, message_id)
424        } else {
425            (String::new(), String::new())
426        };
427        // After the optional `GROUP <g> '<mid>'`, we can have any
428        // combination of `WITH delivery_id = '<id>'` and (for NACK only)
429        // `WITH DELAY <duration>` — in any order. Each clause may appear
430        // at most once. A bare `WITH` followed by neither keyword is a
431        // parse error.
432        let mut delivery_id: Option<String> = None;
433        let mut delay_ms: Option<u64> = None;
434        while self.consume(&Token::With)? {
435            if self.consume_ident_ci("delivery_id")? {
436                if delivery_id.is_some() {
437                    return Err(ParseError::new(
438                        "duplicate WITH delivery_id clause".to_string(),
439                        self.position(),
440                    ));
441                }
442                if !self.consume(&Token::Eq)? {
443                    return Err(ParseError::expected(
444                        vec!["="],
445                        self.peek(),
446                        self.position(),
447                    ));
448                }
449                delivery_id = Some(self.parse_string()?);
450            } else if allow_delay && self.consume_ident_ci("DELAY")? {
451                if delay_ms.is_some() {
452                    return Err(ParseError::new(
453                        "duplicate WITH DELAY clause".to_string(),
454                        self.position(),
455                    ));
456                }
457                let value = self.parse_float()?;
458                let unit = self.parse_queue_duration_unit()?;
459                delay_ms = Some((value * unit).max(0.0) as u64);
460            } else {
461                let mut expected = vec!["delivery_id"];
462                if allow_delay {
463                    expected.push("DELAY");
464                }
465                return Err(ParseError::expected(expected, self.peek(), self.position()));
466            }
467        }
468        if group.is_empty() && delivery_id.is_none() {
469            let mut expected = vec!["GROUP", "WITH delivery_id"];
470            if allow_delay {
471                expected.push("WITH DELAY");
472            }
473            return Err(ParseError::expected(expected, self.peek(), self.position()));
474        }
475        Ok((group, message_id, delivery_id, delay_ms))
476    }
477
478    /// Parse the optional `PRIORITY <int>`, `KEY '<key>'`, `DEDUP '<key>'`,
479    /// `DELAY <duration>`, and `AVAILABLE AT <unix_ms>` tail of a
480    /// `QUEUE PUSH` / `RPUSH` / `LPUSH`
481    /// (issue #722). Clauses may appear in any order; each may appear at
482    /// most once. `AVAILABLE AT` and `DELAY` are mutually exclusive — both
483    /// produce a per-message availability instant, so accepting both
484    /// would force the parser to pick a winner silently.
485    fn parse_push_tail_clauses(
486        &mut self,
487    ) -> Result<
488        (
489            Option<i32>,
490            Option<String>,
491            Option<QueueAvailability>,
492            Option<String>,
493        ),
494        ParseError,
495    > {
496        let mut priority: Option<i32> = None;
497        let mut key: Option<String> = None;
498        let mut available: Option<QueueAvailability> = None;
499        let mut dedup_key: Option<String> = None;
500        loop {
501            if self.consume(&Token::Priority)? {
502                if priority.is_some() {
503                    return Err(ParseError::new(
504                        "duplicate PRIORITY clause".to_string(),
505                        self.position(),
506                    ));
507                }
508                priority = Some(self.parse_integer()? as i32);
509            } else if self.consume(&Token::Key)? {
510                if key.is_some() {
511                    return Err(ParseError::new(
512                        "duplicate KEY clause".to_string(),
513                        self.position(),
514                    ));
515                }
516                if available.is_some() {
517                    return Err(ParseError::new(
518                        "QUEUE PUSH KEY cannot be combined with DELAY / AVAILABLE AT".to_string(),
519                        self.position(),
520                    ));
521                }
522                key = Some(self.parse_string()?);
523            } else if self.consume_ident_ci("DEDUP")? {
524                if dedup_key.is_some() {
525                    return Err(ParseError::new(
526                        "duplicate DEDUP clause".to_string(),
527                        self.position(),
528                    ));
529                }
530                dedup_key = Some(self.parse_string()?);
531            } else if self.peek_ident_ci("DELAY") {
532                self.advance()?;
533                if available.is_some() {
534                    return Err(ParseError::new(
535                        "QUEUE PUSH accepts at most one of DELAY / AVAILABLE AT".to_string(),
536                        self.position(),
537                    ));
538                }
539                if key.is_some() {
540                    return Err(ParseError::new(
541                        "QUEUE PUSH KEY cannot be combined with DELAY / AVAILABLE AT".to_string(),
542                        self.position(),
543                    ));
544                }
545                let value = self.parse_float()?;
546                let unit = self.parse_queue_duration_unit()?;
547                available = Some(QueueAvailability::DelayMs((value * unit).max(0.0) as u64));
548            } else if self.peek_ident_ci("AVAILABLE") {
549                self.advance()?;
550                if !self.consume_ident_ci("AT")? {
551                    return Err(ParseError::expected(
552                        vec!["AT"],
553                        self.peek(),
554                        self.position(),
555                    ));
556                }
557                if available.is_some() {
558                    return Err(ParseError::new(
559                        "QUEUE PUSH accepts at most one of DELAY / AVAILABLE AT".to_string(),
560                        self.position(),
561                    ));
562                }
563                if key.is_some() {
564                    return Err(ParseError::new(
565                        "QUEUE PUSH KEY cannot be combined with DELAY / AVAILABLE AT".to_string(),
566                        self.position(),
567                    ));
568                }
569                let unix_ms = self.parse_integer()?.max(0) as u64;
570                available = Some(QueueAvailability::AtUnixMs(unix_ms));
571            } else {
572                break;
573            }
574        }
575        Ok((priority, key, available, dedup_key))
576    }
577
578    /// Parse an optional `WAIT <duration>` tail (slice A of PRD #718).
579    ///
580    /// Returns `None` if the next token is not the `WAIT` keyword. When
581    /// `WAIT` is present, a duration literal is mandatory — a bare
582    /// `WAIT` with no number errors with a clear "expected: number"
583    /// message bubbled up from `parse_float`.
584    fn parse_optional_wait_clause(&mut self) -> Result<Option<u64>, ParseError> {
585        if !self.peek_ident_ci("WAIT") {
586            return Ok(None);
587        }
588        self.advance()?;
589        let value = self.parse_float()?;
590        let unit = self.parse_queue_duration_unit()?;
591        Ok(Some((value * unit).max(0.0) as u64))
592    }
593
594    /// Reject a stray `WAIT …` tail after a command form that does not
595    /// support blocking reads (`POP`, `PEEK`). The error names the
596    /// command so the operator sees why their statement was refused.
597    fn reject_wait_clause(&mut self, command: &'static str) -> Result<(), ParseError> {
598        if self.peek_ident_ci("WAIT") {
599            return Err(ParseError::new(
600                format!("QUEUE {command} does not support WAIT; use QUEUE READ … WAIT <duration>"),
601                self.position(),
602            ));
603        }
604        Ok(())
605    }
606
607    fn peek_ident_ci(&self, name: &str) -> bool {
608        matches!(self.peek(), Token::Ident(id) if id.eq_ignore_ascii_case(name))
609    }
610
611    fn consume_queue_mode(&mut self) -> Result<Option<QueueMode>, ParseError> {
612        match self.peek() {
613            Token::Work => {
614                self.advance()?;
615                Ok(Some(QueueMode::Work))
616            }
617            Token::Ident(name) => {
618                if let Some(mode) = QueueMode::parse(name) {
619                    self.advance()?;
620                    Ok(Some(mode))
621                } else {
622                    Ok(None)
623                }
624            }
625            _ => Ok(None),
626        }
627    }
628
629    fn parse_queue_mode(&mut self) -> Result<QueueMode, ParseError> {
630        match self.consume_queue_mode()? {
631            Some(mode) => Ok(mode),
632            None => Err(ParseError::expected(
633                vec!["FANOUT", "WORK", "STANDARD", "FIFO"],
634                self.peek(),
635                self.position(),
636            )),
637        }
638    }
639
640    /// Parse duration unit for queue TTL
641    fn parse_queue_duration_unit(&mut self) -> Result<f64, ParseError> {
642        match self.peek().clone() {
643            Token::Min => {
644                self.advance()?;
645                Ok(60_000.0)
646            }
647            Token::Ident(ref unit) => {
648                let mult = match unit.to_ascii_lowercase().as_str() {
649                    "ms" => 1.0,
650                    "s" | "sec" | "secs" => 1_000.0,
651                    "m" | "min" | "mins" => 60_000.0,
652                    "h" | "hr" | "hrs" => 3_600_000.0,
653                    "d" | "day" | "days" => 86_400_000.0,
654                    _ => return Ok(1_000.0),
655                };
656                self.advance()?;
657                Ok(mult)
658            }
659            _ => Ok(1_000.0),
660        }
661    }
662}
reddb_rql/parser/queue.rs

reddb_rql/parser/
queue.rs
