grammers_session/storages/
sqlite.rs

1// Copyright 2020 - developers of the `grammers` project.
2//
3// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
4// https://www.apache.org/licenses/LICENSE-2.0> or the MIT license
5// <LICENSE-MIT or https://opensource.org/licenses/MIT>, at your
6// option. This file may not be copied, modified, or distributed
7// except according to those terms.
8
9use std::collections::HashMap;
10use std::fmt;
11use std::net::AddrParseError;
12use std::path::Path;
13use std::sync::Mutex;
14
15use libsql::{named_params, params};
16use tokio::sync::Mutex as AsyncMutex;
17
18use crate::types::{
19    ChannelKind, ChannelState, DcOption, PeerAuth, PeerId, PeerInfo, PeerKind, UpdateState,
20    UpdatesState,
21};
22use crate::{BoxFuture, DEFAULT_DC, KNOWN_DC_OPTIONS, Session};
23
24const VERSION: i64 = 1;
25
26struct Database(libsql::Connection);
27
28struct Cache {
29    pub home_dc: i32,
30    pub dc_options: HashMap<i32, DcOption>,
31}
32
33/// SQLite-based storage. This is the recommended option.
34pub struct SqliteSession {
35    database: AsyncMutex<Database>,
36    cache: Mutex<Cache>,
37}
38
39#[derive(Debug)]
40pub enum SqliteSessionError {
41    Poisoned,
42    AddrParse(std::net::AddrParseError),
43    Sql(libsql::Error),
44    InvalidAuthKeyLength(usize),
45}
46
47impl std::error::Error for SqliteSessionError {}
48
49impl fmt::Display for SqliteSessionError {
50    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
51        match self {
52            SqliteSessionError::Poisoned => write!(f, "session lock is poisoned"),
53            SqliteSessionError::AddrParse(_) => write!(f, "invalid socket address syntax"),
54            SqliteSessionError::Sql(err) => write!(f, "{err}"),
55            SqliteSessionError::InvalidAuthKeyLength(actual) => {
56                write!(f, "invalid auth_key length: expected 256, got {actual}")
57            }
58        }
59    }
60}
61
62impl From<AddrParseError> for SqliteSessionError {
63    fn from(x: AddrParseError) -> Self {
64        Self::AddrParse(x)
65    }
66}
67
68impl From<libsql::Error> for SqliteSessionError {
69    fn from(x: libsql::Error) -> Self {
70        Self::Sql(x)
71    }
72}
73
74#[repr(u8)]
75enum PeerSubtype {
76    UserSelf = 1,
77    UserBot = 2,
78    UserSelfBot = 3,
79    Megagroup = 4,
80    Broadcast = 8,
81    Gigagroup = 12,
82    Community = 16,
83}
84
85impl Database {
86    async fn init(&self) -> libsql::Result<()> {
87        let mut user_version: i64 = self
88            .fetch_one("PRAGMA user_version", params![], |row| row.get(0))
89            .await?
90            .unwrap_or(0);
91        if user_version == VERSION {
92            return Ok(());
93        }
94
95        if user_version == 0 {
96            self.migrate_v0_to_v1().await?;
97            user_version += 1;
98        }
99        if user_version == VERSION {
100            // Can't bind PRAGMA parameters, but `VERSION` is not user-controlled input.
101            self.0
102                .execute(&format!("PRAGMA user_version = {VERSION}"), params![])
103                .await?;
104        }
105        Ok(())
106    }
107
108    async fn migrate_v0_to_v1(&self) -> libsql::Result<()> {
109        let transaction = self.begin_transaction().await?;
110        transaction
111            .execute(
112                "CREATE TABLE dc_home (
113                dc_id INTEGER NOT NULL,
114                PRIMARY KEY(dc_id))",
115                params![],
116            )
117            .await?;
118        transaction
119            .execute(
120                "CREATE TABLE dc_option (
121                dc_id INTEGER NOT NULL,
122                ipv4 TEXT NOT NULL,
123                ipv6 TEXT NOT NULL,
124                auth_key BLOB,
125                PRIMARY KEY (dc_id))",
126                params![],
127            )
128            .await?;
129        transaction
130            .execute(
131                "CREATE TABLE peer_info (
132                peer_id INTEGER NOT NULL,
133                hash INTEGER,
134                subtype INTEGER,
135                PRIMARY KEY (peer_id))",
136                params![],
137            )
138            .await?;
139        transaction
140            .execute(
141                "CREATE TABLE update_state (
142                pts INTEGER NOT NULL,
143                qts INTEGER NOT NULL,
144                date INTEGER NOT NULL,
145                seq INTEGER NOT NULL)",
146                params![],
147            )
148            .await?;
149        transaction
150            .execute(
151                "CREATE TABLE channel_state (
152                peer_id INTEGER NOT NULL,
153                pts INTEGER NOT NULL,
154                PRIMARY KEY (peer_id))",
155                params![],
156            )
157            .await?;
158
159        transaction.commit().await?;
160        Ok(())
161    }
162
163    async fn begin_transaction(&self) -> libsql::Result<libsql::Transaction> {
164        self.0.transaction().await
165    }
166
167    async fn fetch_one<
168        T,
169        P: libsql::params::IntoParams,
170        F: FnOnce(libsql::Row) -> libsql::Result<T>,
171    >(
172        &self,
173        statement: &str,
174        params: P,
175        select: F,
176    ) -> libsql::Result<Option<T>> {
177        let mut statement = self.0.prepare(statement).await?;
178        let result = statement.query_row(params).await;
179        match result {
180            Ok(value) => Ok(Some(select(value)?)),
181            Err(libsql::Error::QueryReturnedNoRows) => Ok(None),
182            Err(e) => Err(e),
183        }
184    }
185
186    async fn fetch_all<
187        T,
188        P: libsql::params::IntoParams,
189        F: FnMut(libsql::Row) -> Result<T, SqliteSessionError>,
190    >(
191        &self,
192        statement: &str,
193        params: P,
194        mut select: F,
195    ) -> Result<Vec<T>, SqliteSessionError> {
196        let statement = self.0.prepare(statement).await?;
197        let mut rows = statement.query(params).await?;
198        let mut result = Vec::new();
199        while let Some(row) = rows.next().await? {
200            result.push(select(row)?);
201        }
202        Ok(result)
203    }
204}
205
206impl SqliteSession {
207    /// Open a connection to the SQLite database at `path`,
208    /// creating one if it doesn't exist.
209    pub async fn open<P: AsRef<Path>>(path: P) -> Result<Self, SqliteSessionError> {
210        let conn = libsql::Builder::new_local(path).build().await?.connect()?;
211        let db = Database(conn);
212        db.init().await?;
213
214        let home_dc = db
215            .fetch_one("SELECT * FROM dc_home LIMIT 1", named_params![], |row| {
216                row.get::<i32>(0)
217            })
218            .await?
219            .unwrap_or(DEFAULT_DC);
220
221        let dc_options = db
222            .fetch_all("SELECT * FROM dc_option", named_params![], |row| {
223                Ok(DcOption {
224                    id: row.get::<i32>(0)?,
225                    ipv4: row.get::<String>(1)?.parse()?,
226                    ipv6: row.get::<String>(2)?.parse()?,
227                    auth_key: match row.get::<Option<Vec<u8>>>(3)? {
228                        None => None,
229                        Some(auth_key) => Some(auth_key.try_into().map_err(|v: Vec<u8>| {
230                            SqliteSessionError::InvalidAuthKeyLength(v.len())
231                        })?),
232                    },
233                })
234            })
235            .await?
236            .into_iter()
237            .map(|dc_option| (dc_option.id, dc_option))
238            .collect();
239
240        Ok(SqliteSession {
241            database: AsyncMutex::new(db),
242            cache: Mutex::new(Cache {
243                home_dc,
244                dc_options,
245            }),
246        })
247    }
248}
249
250impl Session for SqliteSession {
251    type Error = SqliteSessionError;
252
253    fn home_dc_id(&self) -> Result<i32, SqliteSessionError> {
254        Ok(self
255            .cache
256            .lock()
257            .map_err(|_| SqliteSessionError::Poisoned)?
258            .home_dc)
259    }
260
261    fn set_home_dc_id(&self, dc_id: i32) -> BoxFuture<'_, Result<(), SqliteSessionError>> {
262        let ok = match self.cache.lock() {
263            Err(_) => Err(SqliteSessionError::Poisoned),
264            Ok(mut x) => {
265                x.home_dc = dc_id;
266                Ok(())
267            }
268        };
269        Box::pin(async move {
270            ok?;
271
272            let transaction = self.database.lock().await.begin_transaction().await?;
273            transaction
274                .execute("DELETE FROM dc_home", params![])
275                .await?;
276            let stmt = transaction
277                .prepare("INSERT INTO dc_home VALUES (:dc_id)")
278                .await?;
279            stmt.execute(named_params! {":dc_id": dc_id}).await?;
280            transaction.commit().await?;
281            Ok(())
282        })
283    }
284
285    fn dc_option(&self, dc_id: i32) -> Result<Option<DcOption>, SqliteSessionError> {
286        Ok(self
287            .cache
288            .lock()
289            .map_err(|_| SqliteSessionError::Poisoned)?
290            .dc_options
291            .get(&dc_id)
292            .cloned()
293            .or_else(|| {
294                KNOWN_DC_OPTIONS
295                    .iter()
296                    .find(|dc_option| dc_option.id == dc_id)
297                    .cloned()
298            }))
299    }
300
301    fn set_dc_option(&self, dc_option: &DcOption) -> BoxFuture<'_, Result<(), SqliteSessionError>> {
302        let ok = match self.cache.lock() {
303            Err(_) => Err(SqliteSessionError::Poisoned),
304            Ok(mut x) => {
305                x.dc_options.insert(dc_option.id, dc_option.clone());
306                Ok(())
307            }
308        };
309
310        let dc_option = dc_option.clone();
311        Box::pin(async move {
312            ok?;
313
314            let db = self.database.lock().await;
315            db.0.execute(
316                "INSERT OR REPLACE INTO dc_option VALUES (:dc_id, :ipv4, :ipv6, :auth_key)",
317                named_params! {
318                    ":dc_id": dc_option.id,
319                    ":ipv4": dc_option.ipv4.to_string(),
320                    ":ipv6": dc_option.ipv6.to_string(),
321                    ":auth_key": dc_option.auth_key.map(|k| k.to_vec()),
322                },
323            )
324            .await?;
325            Ok(())
326        })
327    }
328
329    fn peer(&self, peer: PeerId) -> BoxFuture<'_, Result<Option<PeerInfo>, SqliteSessionError>> {
330        Box::pin(async move {
331            let db = self.database.lock().await;
332            let map_row = |row: libsql::Row| {
333                let subtype = row.get::<Option<i64>>(2)?.map(|s| s as u8);
334                Ok(match peer.kind() {
335                    PeerKind::User => PeerInfo::User {
336                        id: PeerId::user_unchecked(row.get::<i64>(0)?).bare_id_unchecked(),
337                        auth: row.get::<Option<i64>>(1)?.map(PeerAuth::from_hash),
338                        bot: subtype.map(|s| s & PeerSubtype::UserBot as u8 != 0),
339                        is_self: subtype.map(|s| s & PeerSubtype::UserSelf as u8 != 0),
340                    },
341                    PeerKind::Chat => PeerInfo::Chat {
342                        id: peer.bare_id_unchecked(),
343                    },
344                    PeerKind::Channel => PeerInfo::Channel {
345                        id: peer.bare_id_unchecked(),
346                        auth: row.get::<Option<i64>>(1)?.map(PeerAuth::from_hash),
347                        kind: subtype.and_then(|s| {
348                            if (s & PeerSubtype::Gigagroup as u8) == PeerSubtype::Gigagroup as u8 {
349                                Some(ChannelKind::Gigagroup)
350                            } else if s & PeerSubtype::Broadcast as u8 != 0 {
351                                Some(ChannelKind::Broadcast)
352                            } else if s & PeerSubtype::Megagroup as u8 != 0 {
353                                Some(ChannelKind::Megagroup)
354                            } else {
355                                None
356                            }
357                        }),
358                    },
359                })
360            };
361
362            Ok(if let Some(peer_id) = peer.bot_api_dialog_id() {
363                db.fetch_one(
364                    "SELECT * FROM peer_info WHERE peer_id = :peer_id LIMIT 1",
365                    named_params! {":peer_id": peer_id},
366                    map_row,
367                )
368                .await?
369            } else {
370                db.fetch_one(
371                    "SELECT * FROM peer_info WHERE subtype & :type LIMIT 1",
372                    named_params! {":type": PeerSubtype::UserSelf as i64},
373                    map_row,
374                )
375                .await?
376            })
377        })
378    }
379
380    fn cache_peer(&self, peer: PeerInfo) -> BoxFuture<'_, Result<(), SqliteSessionError>> {
381        Box::pin(async move {
382            let peer = if let Some(mut existing_peer) = self.peer(peer.id()).await? {
383                existing_peer.extend_info(&peer);
384                existing_peer
385            } else {
386                peer
387            };
388
389            let db = self.database.lock().await;
390            let stmt =
391                db.0.prepare("INSERT OR REPLACE INTO peer_info VALUES (:peer_id, :hash, :subtype)")
392                    .await?;
393            let subtype = match peer {
394                PeerInfo::User { bot, is_self, .. } => {
395                    match (bot.unwrap_or_default(), is_self.unwrap_or_default()) {
396                        (true, true) => Some(PeerSubtype::UserSelfBot),
397                        (true, false) => Some(PeerSubtype::UserBot),
398                        (false, true) => Some(PeerSubtype::UserSelf),
399                        (false, false) => None,
400                    }
401                }
402                PeerInfo::Chat { .. } => None,
403                PeerInfo::Channel { kind, .. } => kind.map(|kind| match kind {
404                    ChannelKind::Megagroup => PeerSubtype::Megagroup,
405                    ChannelKind::Broadcast => PeerSubtype::Broadcast,
406                    ChannelKind::Gigagroup => PeerSubtype::Gigagroup,
407                    ChannelKind::Community => PeerSubtype::Community,
408                }),
409            };
410            let mut params = vec![];
411            let peer_id = peer.id().bot_api_dialog_id_unchecked();
412            params.push((":peer_id".to_owned(), peer_id));
413            let hash = peer.auth().unwrap_or_default().hash();
414            if peer.auth().is_some() {
415                params.push((":hash".to_owned(), hash));
416            }
417            if let Some(subtype) = subtype.map(|s| s as i64) {
418                params.push((":subtype".to_owned(), subtype));
419            }
420            stmt.execute(params).await?;
421            Ok(())
422        })
423    }
424
425    fn cache_peers(&self, peers: Vec<PeerInfo>) -> BoxFuture<'_, Result<(), SqliteSessionError>> {
426        Box::pin(async move {
427            let mut result = Ok(());
428            let mut extended_peers = Vec::with_capacity(peers.len());
429            for peer in peers {
430                extended_peers.push(match self.peer(peer.id()).await {
431                    Ok(Some(mut existing_peer)) => {
432                        existing_peer.extend_info(&peer);
433                        existing_peer
434                    }
435                    Ok(None) => peer,
436                    Err(e) => {
437                        result = Err(e);
438                        peer
439                    }
440                });
441            }
442
443            let db = self.database.lock().await;
444            let tx = db.0.transaction().await.unwrap();
445            let stmt = tx
446                .prepare("INSERT OR REPLACE INTO peer_info VALUES (:peer_id, :hash, :subtype)")
447                .await
448                .unwrap();
449            for peer in extended_peers {
450                let subtype = match peer {
451                    PeerInfo::User { bot, is_self, .. } => {
452                        match (bot.unwrap_or_default(), is_self.unwrap_or_default()) {
453                            (true, true) => Some(PeerSubtype::UserSelfBot),
454                            (true, false) => Some(PeerSubtype::UserBot),
455                            (false, true) => Some(PeerSubtype::UserSelf),
456                            (false, false) => None,
457                        }
458                    }
459                    PeerInfo::Chat { .. } => None,
460                    PeerInfo::Channel { kind, .. } => kind.map(|kind| match kind {
461                        ChannelKind::Megagroup => PeerSubtype::Megagroup,
462                        ChannelKind::Broadcast => PeerSubtype::Broadcast,
463                        ChannelKind::Gigagroup => PeerSubtype::Gigagroup,
464                        ChannelKind::Community => PeerSubtype::Community,
465                    }),
466                };
467                let mut params = vec![];
468                let peer_id = peer.id().bot_api_dialog_id_unchecked();
469                params.push((":peer_id".to_owned(), Some(peer_id)));
470                params.push((":hash".to_owned(), peer.auth().map(|a| a.hash())));
471                params.push((":subtype".to_owned(), subtype.map(|s| s as i64)));
472                if let Err(err) = stmt.execute(params).await {
473                    result = Err(err.into());
474                }
475                stmt.reset();
476            }
477            tx.commit().await?;
478            result
479        })
480    }
481
482    fn updates_state(&self) -> BoxFuture<'_, Result<UpdatesState, SqliteSessionError>> {
483        Box::pin(async move {
484            let db = self.database.lock().await;
485            let mut state = db
486                .fetch_one(
487                    "SELECT * FROM update_state LIMIT 1",
488                    named_params![],
489                    |row| {
490                        Ok(UpdatesState {
491                            pts: row.get(0)?,
492                            qts: row.get(1)?,
493                            date: row.get(2)?,
494                            seq: row.get(3)?,
495                            channels: Vec::new(),
496                        })
497                    },
498                )
499                .await?
500                .unwrap_or_default();
501            state.channels = db
502                .fetch_all("SELECT * FROM channel_state", named_params![], |row| {
503                    Ok(ChannelState {
504                        id: row.get(0)?,
505                        pts: row.get(1)?,
506                    })
507                })
508                .await?;
509            Ok(state)
510        })
511    }
512
513    fn set_update_state(
514        &self,
515        update: UpdateState,
516    ) -> BoxFuture<'_, Result<(), SqliteSessionError>> {
517        Box::pin(async move {
518            let db = self.database.lock().await;
519            let transaction = db.begin_transaction().await?;
520
521            match update {
522                UpdateState::All(updates_state) => {
523                    transaction
524                        .execute("DELETE FROM update_state", params![])
525                        .await?;
526                    transaction
527                        .execute(
528                            "INSERT INTO update_state VALUES (:pts, :qts, :date, :seq)",
529                            named_params! {
530                                ":pts": updates_state.pts,
531                                ":qts": updates_state.qts,
532                                ":date": updates_state.date,
533                                ":seq": updates_state.seq,
534                            },
535                        )
536                        .await?;
537
538                    transaction
539                        .execute("DELETE FROM channel_state", params![])
540                        .await?;
541                    for channel in updates_state.channels {
542                        transaction
543                            .execute(
544                                "INSERT INTO channel_state VALUES (:peer_id, :pts)",
545                                named_params! {
546                                    ":peer_id": channel.id,
547                                    ":pts": channel.pts,
548                                },
549                            )
550                            .await?;
551                    }
552                }
553                UpdateState::Primary { pts, date, seq } => {
554                    let previous = db
555                        .fetch_one(
556                            "SELECT * FROM update_state LIMIT 1",
557                            named_params![],
558                            |_| Ok(()),
559                        )
560                        .await?;
561
562                    if previous.is_some() {
563                        transaction
564                            .execute(
565                                "UPDATE update_state SET pts = :pts, date = :date, seq = :seq",
566                                named_params! {
567                                    ":pts": pts,
568                                    ":date": date,
569                                    ":seq": seq,
570                                },
571                            )
572                            .await?;
573                    } else {
574                        transaction
575                            .execute(
576                                "INSERT INTO update_state VALUES (:pts, 0, :date, :seq)",
577                                named_params! {
578                                    ":pts": pts,
579                                    ":date": date,
580                                    ":seq": seq,
581                                },
582                            )
583                            .await?;
584                    }
585                }
586                UpdateState::Secondary { qts } => {
587                    let previous = db
588                        .fetch_one(
589                            "SELECT * FROM update_state LIMIT 1",
590                            named_params![],
591                            |_| Ok(()),
592                        )
593                        .await?;
594
595                    if previous.is_some() {
596                        transaction
597                            .execute(
598                                "UPDATE update_state SET qts = :qts",
599                                named_params! {":qts": qts},
600                            )
601                            .await?;
602                    } else {
603                        transaction
604                            .execute(
605                                "INSERT INTO update_state VALUES (0, :qts, 0, 0)",
606                                named_params! {":qts": qts},
607                            )
608                            .await?;
609                    }
610                }
611                UpdateState::Channel { id, pts } => {
612                    transaction
613                        .execute(
614                            "INSERT OR REPLACE INTO channel_state VALUES (:peer_id, :pts)",
615                            named_params! {
616                                ":peer_id": id,
617                                ":pts": pts,
618                            },
619                        )
620                        .await?;
621                }
622            }
623
624            transaction.commit().await?;
625            Ok(())
626        })
627    }
628}
629
630#[cfg(test)]
631mod tests {
632    use std::net::{Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6};
633
634    use {DcOption, KNOWN_DC_OPTIONS, PeerInfo, Session, UpdateState};
635
636    use super::*;
637
638    #[test]
639    fn exercise_sqlite_session() {
640        tokio::runtime::Builder::new_current_thread()
641            .enable_all()
642            .build()
643            .unwrap()
644            .block_on(do_exercise_sqlite_session());
645    }
646
647    async fn do_exercise_sqlite_session() {
648        let session = SqliteSession::open(":memory:").await.unwrap();
649
650        assert_eq!(session.home_dc_id().unwrap(), DEFAULT_DC);
651        session.set_home_dc_id(DEFAULT_DC + 1).await.unwrap();
652        assert_eq!(session.home_dc_id().unwrap(), DEFAULT_DC + 1);
653
654        assert_eq!(
655            session.dc_option(KNOWN_DC_OPTIONS[0].id).unwrap(),
656            Some(KNOWN_DC_OPTIONS[0].clone())
657        );
658        let new_dc_option = DcOption {
659            id: KNOWN_DC_OPTIONS
660                .iter()
661                .map(|dc_option| dc_option.id)
662                .max()
663                .unwrap()
664                + 1,
665            ipv4: SocketAddrV4::new(Ipv4Addr::from_bits(0), 1),
666            ipv6: SocketAddrV6::new(Ipv6Addr::from_bits(0), 1, 0, 0),
667            auth_key: Some([1; 256]),
668        };
669        assert_eq!(session.dc_option(new_dc_option.id).unwrap(), None);
670        session.set_dc_option(&new_dc_option).await.unwrap();
671        assert_eq!(
672            session.dc_option(new_dc_option.id).unwrap(),
673            Some(new_dc_option)
674        );
675
676        assert_eq!(session.peer(PeerId::self_user()).await.unwrap(), None);
677        assert_eq!(session.peer(PeerId::user_unchecked(1)).await.unwrap(), None);
678        let peer = PeerInfo::User {
679            id: 1,
680            auth: None,
681            bot: Some(true),
682            is_self: Some(true),
683        };
684        session.cache_peer(peer.clone()).await.unwrap();
685        assert_eq!(
686            session.peer(PeerId::self_user()).await.unwrap(),
687            Some(peer.clone())
688        );
689        assert_eq!(
690            session.peer(PeerId::user_unchecked(1)).await.unwrap(),
691            Some(peer)
692        );
693
694        assert_eq!(
695            session.peer(PeerId::channel_unchecked(1)).await.unwrap(),
696            None
697        );
698        let peer = PeerInfo::Channel {
699            id: 1,
700            auth: Some(PeerAuth::from_hash(-1)),
701            kind: Some(ChannelKind::Broadcast),
702        };
703        session.cache_peer(peer.clone()).await.unwrap();
704        assert_eq!(
705            session.peer(PeerId::channel_unchecked(1)).await.unwrap(),
706            Some(peer)
707        );
708
709        let peers = vec![
710            PeerInfo::Channel {
711                id: 2,
712                auth: Some(PeerAuth::from_hash(-2)),
713                kind: Some(ChannelKind::Megagroup),
714            },
715            PeerInfo::Channel {
716                id: 3,
717                auth: Some(PeerAuth::from_hash(-3)),
718                kind: None,
719            },
720        ];
721        session.cache_peers(peers.clone()).await.unwrap();
722        for peer in peers.iter() {
723            assert_eq!(session.peer(peer.id()).await.unwrap().as_ref(), Some(peer));
724        }
725
726        // If the inserted peer is going to be updated with some fields changed
727        // from `Some` to `None`, the old `Some` is expected to be kept.
728        let new_inserted_peers = vec![
729            PeerInfo::Channel {
730                id: 2,
731                auth: None,
732                kind: Some(ChannelKind::Broadcast),
733            },
734            PeerInfo::Channel {
735                id: 3,
736                auth: Some(PeerAuth::from_hash(-30)),
737                kind: None,
738            },
739        ];
740        session.cache_peers(new_inserted_peers).await.unwrap();
741        assert_eq!(
742            session.peer(PeerId::channel_unchecked(2)).await.unwrap(),
743            Some(PeerInfo::Channel {
744                id: 2,
745                auth: Some(PeerAuth::from_hash(-2)),
746                kind: Some(ChannelKind::Broadcast),
747            })
748        );
749        assert_eq!(
750            session.peer(PeerId::channel_unchecked(3)).await.unwrap(),
751            Some(PeerInfo::Channel {
752                id: 3,
753                auth: Some(PeerAuth::from_hash(-30)),
754                kind: None
755            })
756        );
757
758        assert_eq!(
759            session.updates_state().await.unwrap(),
760            UpdatesState::default()
761        );
762        session
763            .set_update_state(UpdateState::All(UpdatesState {
764                pts: 1,
765                qts: 2,
766                date: 3,
767                seq: 4,
768                channels: vec![
769                    ChannelState { id: 5, pts: 6 },
770                    ChannelState { id: 7, pts: 8 },
771                ],
772            }))
773            .await
774            .unwrap();
775        session
776            .set_update_state(UpdateState::Primary {
777                pts: 2,
778                date: 4,
779                seq: 5,
780            })
781            .await
782            .unwrap();
783        session
784            .set_update_state(UpdateState::Secondary { qts: 3 })
785            .await
786            .unwrap();
787        session
788            .set_update_state(UpdateState::Channel { id: 7, pts: 9 })
789            .await
790            .unwrap();
791        assert_eq!(
792            session.updates_state().await.unwrap(),
793            UpdatesState {
794                pts: 2,
795                qts: 3,
796                date: 4,
797                seq: 5,
798                channels: vec![
799                    ChannelState { id: 5, pts: 6 },
800                    ChannelState { id: 7, pts: 9 },
801                ],
802            }
803        );
804    }
805}
grammers_session/storages/sqlite.rs

grammers_session/storages/
sqlite.rs
