diff --git a/flake.nix b/flake.nix index c02d699..e6f82b5 100644 --- a/flake.nix +++ b/flake.nix @@ -10,6 +10,7 @@ let pkgs = import nixpkgs { inherit system; }; naersk-lib = pkgs.callPackage naersk { }; + PROJECT_ROOT = builtins.toString ./.; in { defaultPackage = naersk-lib.buildPackage ./.; @@ -42,6 +43,9 @@ cargo-tarpaulin deno]; RUST_SRC_PATH = rustPlatform.rustLibSrc; + shellHook = '' + export PROJECT_ROOT=${PROJECT_ROOT}; + ''; }; }); } diff --git a/migrations/avro/events.ts b/migrations/avro/events.ts index bddb894..c4e1550 100644 --- a/migrations/avro/events.ts +++ b/migrations/avro/events.ts @@ -1,6 +1,7 @@ import {Command} from "https://deno.land/x/cliffy@v0.25.7/command/mod.ts"; import {concat, partial} from 'https://deno.land/x/lodash@4.17.15-es/lodash.js'; -import {join, dirname} from "https://deno.land/std@0.171.0/path/mod.ts"; +import {dirname, join} from "https://deno.land/std@0.171.0/path/mod.ts"; +import avro from "npm:avro-js@1.11.1"; enum AvroPrimitives { @@ -43,7 +44,7 @@ type AvroUnion = AvroTypes[]; type AvroTypes = AvroPrimitives | AvroRecord | AvroUnion | AvroEnum; const CacheChequeEvent: AvroRecord = { - name: "event", + name: "CacheChequeEvent", type: "record", fields: [ {name: "account", type: AvroPrimitives.Long}, @@ -51,7 +52,7 @@ const CacheChequeEvent: AvroRecord = { } const BogusEvent: AvroRecord = { - name: "bogus", + name: "BogusEvent", type: "record", fields: [ {name: "account", type: AvroPrimitives.Long}, @@ -92,7 +93,7 @@ async function drySerializer(outputDir: string, rec: AvroRecord): Promise } async function fileSerializer(outputDir: string, rec: AvroRecord): Promise { - let outputFile = getOutputFileName(outputDir, rec); + const outputFile = getOutputFileName(outputDir, rec); await Deno.mkdir(dirname(outputFile), { recursive: true }); await Deno.writeTextFile(outputFile, JSON.stringify(rec, null, 4)); @@ -102,24 +103,81 @@ function getSerializer(dry: boolean, outputDir: string): Serializer { return dry ? partial(drySerializer, outputDir) : partial(fileSerializer, outputDir); } - -async function main(options: CliOptions): Promise { - let serializer = getSerializer(options.dry, options.output); - for(let e of concat([], allEvents, LedgerEventsSchema)){ - await serializer(e); +async function buildSchemata(options: CliOptions): Promise { + const serializer = getSerializer(options.dry, options.output); + for (const e of concat([], allEvents, LedgerEventsSchema, eventTypes)) { + await serializer(e); } } +async function testShit(): Promise { + const type = avro.parse(CacheChequeEvent); + const buff = await Deno.readFile("/tmp/foo.data"); + console.log(buff); + console.log(type.fromBuffer(buff)); +} + +async function createMockData(): Promise { + const type = avro.parse( + { + "name": "CacheChequeEvent", + "type": "record", + "fields": [ + { + "name": "account", + "type": "long" + } + ] + } + + ); + const e = { + account: 666 + // event_type: CacheChequeEvent.name, + // event: {CacheChequeEvent: {account: 1788301184}} + }; + const encoder = avro.createFileEncoder("/tmp/foo.data", type); + encoder.write({account: 666}); + encoder.end(); + return; + // const file = await Deno.open("/tmp/foo.data", {write: true, create: true}); + // for (let i = 0; i < 1; i++) { + // await Deno.write(file.rid, type.toBuffer(e).buffer); + // } +} + +function buildSchemataCommand() { + return new Command() + .description("Build the avro schemata") + .option("--dry [value:boolean]", "Do not output any files", { + default: false, + }) + .option("--output [value:file]", "directory to output schemata", { + default: '.', + }) + .action(buildSchemata); +} + +function buildMockEventsCommand() { + return new Command() + .description("Generate mock data") + .option("--output [value:file]", "output data file", { + default: '.', + }) + .action(createMockData); +} + +function buildTestingCommand() { + return new Command() + .description("Test mock data") + .action(testShit); + +} await new Command() .name("ledger-avro-schemata-builder") .version("0.1.0") -.description("Build avro schemata for ledger.") -.option("--dry [value:boolean]", "Do not output any files", { - default: false, -}) -.option("--output [value:file]", "directory to output schemata", { - default: '.', -}) -.action(main) +.command('build-schema', buildSchemataCommand()) +.command('create-mock-data', buildMockEventsCommand()) +.command('testing', buildTestingCommand()) .parse(Deno.args); diff --git a/src/event_processor.rs b/src/event_processor.rs index 2b36d42..e0c27c7 100644 --- a/src/event_processor.rs +++ b/src/event_processor.rs @@ -1,41 +1,55 @@ use std::borrow::Borrow; +use std::f32::consts::E; use std::fs; use std::fs::File; +use std::io::{BufReader, Read, Write}; use std::path::PathBuf; +use apache_avro::types::Value; +use apache_avro::Schema::Union; +use apache_avro::{ + from_value, types::Record, Codec, GenericSingleObjectReader, Reader, Schema, Writer, +}; +use resource::resource_str; use rusqlite::Connection; - +use serde::{Deserialize, Serialize}; use crate::util::{CustomResult, Error}; -// struct +#[derive(Deserialize, Serialize, Debug, PartialEq)] +struct CacheChequeEvent { + account: i64, +} + +#[derive(Deserialize, Serialize, Debug, PartialEq)] +enum Event { + CacheCheque(CacheChequeEvent), +} fn process_event_file(conn: &mut Connection, file: File) -> CustomResult<()> { return Ok(()); } pub fn process_event_files(conn: &mut Connection, paths: I) -> CustomResult<()> - where I: Iterator, - I::Item: Borrow { +where + I: Iterator, + I::Item: Borrow, +{ for path in paths { let file = File::open(path.borrow()).map_err(Error::IOError)?; - process_event_file(conn, file).unwrap(); + process_event_file(conn, file)?; } Ok(()) } - mod test { use std::fs::File; + use apache_avro::Schema; use tempfile::tempfile; use super::*; - fn test_basic_logic(file: File) -> Result<(), Error> { - Ok(()) - } - #[test] fn test_basic() -> CustomResult<()> { @@ -44,4 +58,4 @@ mod test { let ret = process_event_file(&mut conn, file); ret } -} \ No newline at end of file +} diff --git a/src/main.rs b/src/main.rs index 552ae7b..a5bdb1f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,7 +1,7 @@ use std::path::PathBuf; -use clap::{Arg, ArgAction, ArgMatches, Command, value_parser}; use clap::parser::ValuesRef; +use clap::{value_parser, Arg, ArgAction, ArgMatches, Command}; use log::{debug, error, info}; use rusqlite::Connection; use serde::Serialize; @@ -10,14 +10,14 @@ use simple_logger::SimpleLogger; use db::securities::list_securities; use crate::db::users::{create_user, list_users}; -use crate::event_processor::process_event_files; +use crate::event_processor::{process_event_files}; use crate::migration::apply_migrations; -mod migration; mod db; +mod event_processor; +mod migration; mod services; mod util; -mod event_processor; #[derive(Debug, Serialize)] struct Output { @@ -27,7 +27,10 @@ struct Output { impl Output { fn success(shit: T) -> Output { - return Output { status: "success".to_string(), result: shit }; + return Output { + status: "success".to_string(), + result: shit, + }; } } @@ -75,8 +78,20 @@ fn new_user_command_line_parser() -> Command { return Command::new("new") .long_flag("new") .about("Create new user.") - .arg(Arg::new("name").long("name").required(true).help("name of new user.").action(ArgAction::Set)) - .arg(Arg::new("public_key").required(true).long("public-key").help("public key of users.").action(ArgAction::Set)); + .arg( + Arg::new("name") + .long("name") + .required(true) + .help("name of new user.") + .action(ArgAction::Set), + ) + .arg( + Arg::new("public_key") + .required(true) + .long("public-key") + .help("public key of users.") + .action(ArgAction::Set), + ); } fn ls_user_command_line_parser() -> Command { @@ -131,19 +146,31 @@ fn command_line_parser() -> Command { .arg_required_else_help(true); } -fn process_new_user_command_line(conn: &mut Connection, m: &ArgMatches, logic: &mut dyn CommandLineUserLogic) { +fn process_new_user_command_line( + conn: &mut Connection, + m: &ArgMatches, + logic: &mut dyn CommandLineUserLogic, +) { debug!("Processing new user command line {:?}.", m); let name = m.get_one::("name").unwrap(); let pk = m.get_one::("public_key").unwrap(); logic.create_user(conn, name, pk); } -fn process_list_user_command_line(conn: &Connection, m: &ArgMatches, logic: &mut dyn CommandLineUserLogic) { +fn process_list_user_command_line( + conn: &Connection, + m: &ArgMatches, + logic: &mut dyn CommandLineUserLogic, +) { debug!("Processing list users command line {:?}.", m); logic.list_user(conn); } -fn process_user_command_line(conn: &mut Connection, m: &ArgMatches, logic: &mut dyn CommandLineUserLogic) { +fn process_user_command_line( + conn: &mut Connection, + m: &ArgMatches, + logic: &mut dyn CommandLineUserLogic, +) { debug!("Processing user command line {:?}.", m); match m.subcommand() { Some(("new", new_user_match)) => { @@ -156,12 +183,20 @@ fn process_user_command_line(conn: &mut Connection, m: &ArgMatches, logic: &mut } } -fn process_list_securities_command_line(conn: &Connection, m: &ArgMatches, logic: &mut dyn CommandLineUserLogic) { +fn process_list_securities_command_line( + conn: &Connection, + m: &ArgMatches, + logic: &mut dyn CommandLineUserLogic, +) { debug!("Processing list securities command line {:?}.", m); logic.list_securities(conn); } -fn process_process_events_command_line(conn: &mut Connection, m: &ArgMatches, logic: &mut dyn CommandLineUserLogic) { +fn process_process_events_command_line( + conn: &mut Connection, + m: &ArgMatches, + logic: &mut dyn CommandLineUserLogic, +) { debug!("Processing process event command line {:?}.", m); let mut files = m.get_many::("PATH").unwrap(); if files.any(|x| !x.exists()) { @@ -172,7 +207,11 @@ fn process_process_events_command_line(conn: &mut Connection, m: &ArgMatches, lo logic.process_events(conn, files); } -fn process_securities_command_line(conn: &Connection, m: &ArgMatches, logic: &mut dyn CommandLineUserLogic) { +fn process_securities_command_line( + conn: &Connection, + m: &ArgMatches, + logic: &mut dyn CommandLineUserLogic, +) { debug!("Processing securities command line {:?}.", m); match m.subcommand() { Some(("ls", list_securities_match)) => { @@ -182,7 +221,11 @@ fn process_securities_command_line(conn: &Connection, m: &ArgMatches, logic: &mu } } -fn process_events_command_line(conn: &mut Connection, m: &ArgMatches, logic: &mut dyn CommandLineUserLogic) { +fn process_events_command_line( + conn: &mut Connection, + m: &ArgMatches, + logic: &mut dyn CommandLineUserLogic, +) { debug!("Processing events command line {:?}.", m); match m.subcommand() { Some(("process", process_events_match)) => { @@ -192,7 +235,11 @@ fn process_events_command_line(conn: &mut Connection, m: &ArgMatches, logic: &mu } } -fn process_command_line(conn: &mut Connection, m: &ArgMatches, logic: &mut dyn CommandLineUserLogic) { +fn process_command_line( + conn: &mut Connection, + m: &ArgMatches, + logic: &mut dyn CommandLineUserLogic, +) { debug!("Processing command line {:?}.", m); match m.subcommand() { Some(("users", user_match)) => { @@ -260,7 +307,6 @@ mod cli_test { assert_eq!(get_verbosity(&m), log::LevelFilter::Info); } - { let cmd = Command::new("test").arg(verbosity_arg()); let m = cmd.get_matches_from(vec!["test"]); @@ -288,7 +334,15 @@ mod cli_test { #[test] fn create_user() { let cmd = command_line_parser(); - let m = cmd.get_matches_from(vec!["test", "users", "new", "--name", "paul", "--public-key", "123abc"]); + let m = cmd.get_matches_from(vec![ + "test", + "users", + "new", + "--name", + "paul", + "--public-key", + "123abc", + ]); let mut logic = TestCommandLineLogic::new(); let mut conn = Connection::open_in_memory().unwrap(); process_command_line(&mut conn, &m, &mut logic); @@ -381,10 +435,12 @@ fn get_connection(m: &ArgMatches) -> Connection { } fn set_logger(m: &ArgMatches) { - SimpleLogger::new().with_level(get_verbosity(&m)).init().unwrap(); + SimpleLogger::new() + .with_level(get_verbosity(&m)) + .init() + .unwrap(); } - fn main() { let command = command_line_parser(); let m = command.get_matches(); @@ -395,4 +451,4 @@ fn main() { apply_migrations(&mut conn).unwrap(); let mut logic = ProCommandLineLogic::new(); process_command_line(&mut conn, &m, &mut logic); -} \ No newline at end of file +}