Merge branch 'fix/antenna-pagination' into 'develop'

fix: antenna pagination

Closes #10535

See merge request firefish/firefish!10553

packages/backend/native-utils/Cargo.lock

@@ -196,6 +196,12 @@ version = "0.21.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "604178f6c5c21f02dc555784810edfb88d34ac2c73b2eae109655649ee73ce3d"
 
+[[package]]
+name = "basen"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1dbe4bb73fd931c4d1aaf53b35d1286c8a948ad00ec92c8e3c856f15fd027f43"
+
 [[package]]
 name = "bigdecimal"
 version = "0.3.1"
@@ -1399,6 +1405,7 @@ name = "native-utils"
 version = "0.0.0"
 dependencies = [
  "async-trait",
+ "basen",
  "cfg-if",
  "chrono",
  "cuid2",
@@ -1410,7 +1417,6 @@ dependencies = [
  "once_cell",
  "parse-display",
  "pretty_assertions",
- "radix_fmt",
  "rand",
  "schemars",
  "sea-orm",
@@ -1831,12 +1837,6 @@ version = "0.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "dc33ff2d4973d518d823d61aa239014831e521c75da58e3df4840d3f47749d09"
 
-[[package]]
-name = "radix_fmt"
-version = "1.0.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ce082a9940a7ace2ad4a8b7d0b1eac6aa378895f18be598230c5f2284ac05426"
-
 [[package]]
 name = "rand"
 version = "0.8.5"

packages/backend/native-utils/Cargo.toml

@@ -31,11 +31,11 @@ serde_json = "1.0.96"
 thiserror = "1.0.40"
 tokio = { version = "1.28.1", features = ["full"] }
 utoipa = "3.3.0"
-radix_fmt = "1.0.0"
 
 # Default enable napi4 feature, see https://nodejs.org/api/n-api.html#node-api-version-matrix
 napi = { version = "2.13.1", default-features = false, features = ["napi6", "tokio_rt"], optional = true }
 napi-derive = { version = "2.12.0", optional = true }
+basen = "0.1.0"
 
 [dev-dependencies]
 pretty_assertions = "1.3.0"

packages/backend/native-utils/migration/Cargo.toml

@@ -10,11 +10,11 @@ path = "src/lib.rs"
 
 [features]
 default = []
-convert = ["dep:native-utils", "dep:indicatif", "dep:futures"]
+convert = ["dep:indicatif", "dep:futures"]
 
 [dependencies]
 serde_json = "1.0.96"
-native-utils = { path = "../", optional = true }
+native-utils = { path = "../" }
 indicatif = { version = "0.17.4", features = ["tokio"], optional = true }
 tokio = { version = "1.28.2", features = ["full"] }
 futures = { version = "0.3.28", optional = true }

packages/backend/native-utils/migration/src/lib.rs

@@ -3,6 +3,7 @@ pub use sea_orm_migration::prelude::*;
 mod m20230531_180824_drop_reversi;
 mod m20230627_185451_index_note_url;
 mod m20230709_000510_move_antenna_to_cache;
+mod m20230806_170616_fix_antenna_stream_ids;
 
 pub struct Migrator;
 
@@ -13,6 +14,7 @@ impl MigratorTrait for Migrator {
             Box::new(m20230531_180824_drop_reversi::Migration),
             Box::new(m20230627_185451_index_note_url::Migration),
             Box::new(m20230709_000510_move_antenna_to_cache::Migration),
+            Box::new(m20230806_170616_fix_antenna_stream_ids::Migration),
         ]
     }
 }

packages/backend/native-utils/migration/src/m20230709_000510_move_antenna_to_cache.rs

@@ -85,7 +85,7 @@ impl MigrationTrait for Migration {
                         )
                         .ignore();
                     }
-                    pipe.query::<()>(&mut redis_conn).unwrap();
+                    pipe.query::<()>(&mut redis_conn).unwrap_or(());
                 }
 
                 let copied = total_num - remaining;

packages/backend/native-utils/migration/src/m20230806_170616_fix_antenna_stream_ids.rs

@@ -0,0 +1,59 @@
+use std::env;
+
+use native_utils::util::id;
+use redis::Commands;
+use sea_orm_migration::prelude::*;
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, _manager: &SchemaManager) -> Result<(), DbErr> {
+        // Replace the sample below with your own migration scripts
+        let cache_url = env::var("CACHE_URL").unwrap();
+        let prefix = env::var("CACHE_PREFIX").unwrap();
+
+        let client = redis::Client::open(cache_url).unwrap();
+        let mut redis_conn = client.get_connection().unwrap();
+
+        let keys: Vec<String> = redis_conn
+            .keys(format!("{}:antennaTimeline:*", prefix))
+            .unwrap();
+        let key_len = keys.len();
+
+        println!(
+            "Fixing corrupted stream IDs: {} timelines to be fixed",
+            key_len
+        );
+
+        for (i, key) in keys.iter().enumerate() {
+            let all_elems: Vec<Vec<Vec<String>>> = redis_conn.xrange_all(key).unwrap(); // Get all post IDs in stream
+            let stream_ids = all_elems
+                .iter()
+                .map(|v| format!("{}-*", id::get_timestamp(&v[1][1]))); // Get correct stream id with timestamp
+            redis_conn.del::<_, ()>(key).unwrap();
+            for (j, v) in stream_ids.enumerate() {
+                redis_conn
+                    .xadd(key, v, &[("note", &all_elems[j][1][1])])
+                    .unwrap_or(());
+            }
+
+            if i % 10 == 0 {
+                println!(
+                    "Fixing streams [{:.2}%]",
+                    (i as f64 / key_len as f64) * 100_f64
+                );
+            }
+        }
+
+        println!("Fixing streams [100.00%]");
+
+        Ok(())
+    }
+
+    async fn down(&self, _manager: &SchemaManager) -> Result<(), DbErr> {
+        // Replace the sample below with your own migration scripts
+        Ok(())
+    }
+}

packages/backend/native-utils/src/util/id.rs

@@ -1,9 +1,9 @@
 //! ID generation utility based on [cuid2]
 
+use basen::BASE36;
 use cfg_if::cfg_if;
 use chrono::Utc;
 use once_cell::sync::OnceCell;
-use radix_fmt::radix_36;
 use std::cmp;
 
 use crate::impl_into_napi_error;
@@ -46,13 +46,21 @@ pub fn create_id(date_num: i64) -> Result<String, ErrorUninitialized> {
             let time = cmp::max(date_num - TIME_2000, 0);
             Ok(format!(
                 "{:0>8}{}",
-                radix_36(time).to_string(),
+                BASE36.encode_var_len(&(time as u64)),
                 gen.create_id()
             ))
         }
     }
 }
 
+pub fn get_timestamp(id: &str) -> i64 {
+    let n: Option<u64> = BASE36.decode_var_len(&id[0..8]);
+    match n {
+        None => -1,
+        Some(n) => n as i64 + TIME_2000,
+    }
+}
+
 cfg_if! {
     if #[cfg(feature = "napi")] {
         use napi_derive::napi;
@@ -68,17 +76,23 @@ cfg_if! {
         pub fn native_create_id(date_num: i64) -> String {
             create_id(date_num).unwrap()
         }
+
+        #[napi]
+        pub fn native_get_timestamp(id: String) -> i64 {
+            get_timestamp(&id)
+        }
     }
 }
 
 #[cfg(test)]
 mod unit_test {
     use crate::util::id;
+    use chrono::Utc;
     use pretty_assertions::{assert_eq, assert_ne};
     use std::thread;
 
     #[test]
-    fn can_generate_unique_ids() {
+    fn can_create_and_decode() {
         assert_eq!(id::create_id(0), Err(id::ErrorUninitialized));
         id::init_id(16, "");
         assert_eq!(id::create_id(0).unwrap().len(), 16);
@@ -86,5 +100,10 @@ mod unit_test {
         let id1 = thread::spawn(|| id::create_id(0).unwrap());
         let id2 = thread::spawn(|| id::create_id(0).unwrap());
         assert_ne!(id1.join().unwrap(), id2.join().unwrap());
+
+        let now = Utc::now().timestamp_millis();
+        let test_id = id::create_id(now).unwrap();
+        let timestamp = id::get_timestamp(&test_id);
+        assert_eq!(now, timestamp);
     }
 }

packages/backend/src/misc/gen-id.ts

@@ -2,6 +2,7 @@ import config from "@/config/index.js";
 import {
 	nativeCreateId,
 	nativeInitIdGenerator,
+	nativeGetTimestamp,
 } from "native-utils/built/index.js";
 
 const length = Math.min(Math.max(config.cuid?.length ?? 16, 16), 24);
@@ -19,3 +20,7 @@ nativeInitIdGenerator(length, fingerprint);
 export function genId(date?: Date): string {
 	return nativeCreateId((date ?? new Date()).getTime());
 }
+
+export function getTimestamp(id: string): number {
+	return nativeGetTimestamp(id);
+}

packages/backend/src/server/api/endpoints/antennas/notes.ts

@@ -2,7 +2,7 @@ import define from "../../define.js";
 import readNote from "@/services/note/read.js";
 import { Antennas, Notes } from "@/models/index.js";
 import { redisClient } from "@/db/redis.js";
-import { genId } from "@/misc/gen-id.js";
+import { genId, getTimestamp } from "@/misc/gen-id.js";
 import { makePaginationQuery } from "../../common/make-pagination-query.js";
 import { generateVisibilityQuery } from "../../common/generate-visibility-query.js";
 import { generateMutedUserQuery } from "../../common/generate-muted-user-query.js";
@@ -61,10 +61,22 @@ export default define(meta, paramDef, async (ps, user) => {
 	}
 
 	const limit = ps.limit + (ps.untilId ? 1 : 0) + (ps.sinceId ? 1 : 0); // untilIdに指定したものも含まれるため+1
+	let end = "+";
+	if (ps.untilDate) {
+		end = ps.untilDate.toString();
+	} else if (ps.untilId) {
+		end = getTimestamp(ps.untilId).toString();
+	}
+	let start = "-";
+	if (ps.sinceDate) {
+		start = ps.sinceDate.toString();
+	} else if (ps.sinceId) {
+		start = getTimestamp(ps.sinceId).toString();
+	}
 	const noteIdsRes = await redisClient.xrevrange(
 		`antennaTimeline:${antenna.id}`,
-		ps.untilDate ?? "+",
-		ps.sinceDate ?? "-",
+		end,
+		start,
 		"COUNT",
 		limit,
 	);

packages/backend/src/services/add-note-to-antenna.ts

@@ -1,6 +1,6 @@
 import type { Antenna } from "@/models/entities/antenna.js";
 import type { Note } from "@/models/entities/note.js";
-import { genId } from "@/misc/gen-id.js";
+import { getTimestamp } from "@/misc/gen-id.js";
 import { redisClient } from "@/db/redis.js";
 import { publishAntennaStream } from "@/services/stream.js";
 import type { User } from "@/models/entities/user.js";
@@ -15,7 +15,7 @@ export async function addNoteToAntenna(
 		"MAXLEN",
 		"~",
 		"200",
-		"*",
+		`${getTimestamp(note.id)}-*`,
 		"note",
 		note.id,
 	);