Skip to content

WIP: migrations #241

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
jsparber wants to merge 1 commit into
base: main
Choose a base branch
from
+145 −0
Draft

WIP: migrations #241

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
142 changes: 142 additions & 0 deletions reflection-node/src/migrations.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,142 @@
add `_reflection_migrations` table to store version, checksum, execution time, filename

struct Migrations {};

#[derive(Debug, FromRow)]
struct Operation {
#[sqlx(rename = "log_id")]
pub log_id_hash: String,
// FIXME: needs to be hex decoded hex::decode() and we need to hope that the public key format didn't change
//pub public_key: String,
pub body: Option<Vec<u8>>,
}

#[derive(Clone, Debug, PartialEq, Eq, StdHash, Serialize, Deserialize)]
pub struct LogId(LogType, TopicId);

impl LogId {
pub fn new(log_type: LogType, topic: &TopicId) -> Self {
Self(log_type, *topic)
}

pub fn hash(&mut self) -> u64 {
let mut s = DefaultHasher::new();
t.hash(&mut s);
s.finish()
}
}

PublicKey::from_str(&row.public_key).unwrap(),


impl Migrations {
let connection_options = sqlx::sqlite::SqliteConnectOptions::new()
.shared_cache(true)
.create_if_missing(true);

let connection_options = connection_options.filename(db_file);
let db_file = sqlx::sqlite::SqlitePool::connect_with(connection_options).await?

async fn load_operations(
&self,
public_key: &PublicKey,
) -> Result<Vec<Operation>, SqliteStoreError> {
let operations = query_as::<_, Operation>(
"
SELECT
log_id,
body,
FROM
operations_v1
GROUP_BY
log_id
",
)
.bind(public_key.to_string())
.fetch_all(&self.pool)
.await?;
}


fn calculate_hash<T: StdHash>(t: &T) -> u64 {
let mut s = DefaultHasher::new();
t.hash(&mut s);
s.finish()
}

pub async fn operations_for_topic(
&self,
operation_store: &OperationStore,
id: &TopicId,
) -> sqlx::Result<Vec<(PublicKey, Option<Vec<u8>>)>> {
let topics = todo!("get all topics");

topcis.map(|id|
(id,
LogId::new(LogType::Delta, id).hash(),
LogId::new(LogType::Snapshot, id).hash(),
));

let operations = self.load_operations();

Ok(result)
}


/*
async fn get_log(
&self,
public_key: &PublicKey,
log_id: &LogId,
from: Option<u64>,
) -> Result<Option<Vec<(Header<Extensions>, Option<Body>)>>, SqliteStoreError> {
let operations = query_as::<_, OperationRow>(
"
SELECT
hash,
log_id,
version,
public_key,
signature,
payload_size,
payload_hash,
timestamp,
seq_num,
backlink,
previous,
extensions,
body,
header_bytes
FROM
operations_v1
WHERE
public_key = ?
AND log_id = ?
AND CAST(seq_num AS NUMERIC) >= CAST(? as NUMERIC)
ORDER BY
CAST(seq_num AS NUMERIC)
",
)
.bind(public_key.to_string())
.bind(calculate_hash(log_id).to_string())
.bind(from.unwrap_or(0).to_string())
.fetch_all(&self.pool)
.await?;

let log: Vec<(Header<E>, Option<Body>)> = operations
.into_iter()
.map(|operation| {
(
operation.clone().into(),
operation.body.map(|body| body.into()),
)
})
.collect();

if log.is_empty() {
Ok(None)
} else {
Ok(Some(log))
}
}
*/
3 changes: 3 additions & 0 deletions reflection-node/src/node.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -116,6 +116,9 @@ impl Node {
})
}

pub async fn run_migrations(&self) {
}

pub async fn set_connection_mode(
&self,
connection_mode: ConnectionMode,
Expand Down
Loading