From 0756e9995934d6c80c4011b0efcc1f46b3f54cd5 Mon Sep 17 00:00:00 2001 From: Filipriec Date: Fri, 21 Aug 2026 22:36:41 +0200 Subject: [PATCH] bulk import --- common/proto/tables_data.proto | 58 ++- common/src/proto/descriptor.bin | Bin 186516 -> 189349 bytes common/src/proto/komp_ac.tables_data.rs | 396 +++++++++++++++++- web/locales/cs/main.ftl | 5 +- web/locales/en/main.ftl | 5 +- web/locales/sk/main.ftl | 5 +- web/src/pages/import_export/import/logic.rs | 141 +++++-- .../pages/import_export/import/progress.rs | 12 +- web/src/pages/import_export/import/ui.rs | 73 ++-- 9 files changed, 600 insertions(+), 95 deletions(-) diff --git a/common/proto/tables_data.proto b/common/proto/tables_data.proto index 2be621c3..ed69872b 100644 --- a/common/proto/tables_data.proto +++ b/common/proto/tables_data.proto @@ -29,7 +29,7 @@ service TablesData { rpc PostAccountingTableData(PostAccountingTableDataRequest) returns (PostTableDataResponse); - // Insert multiple rows by applying PostTableData behavior to each row. + // Insert multiple rows atomically by applying PostTableData behavior to each row. // // Behavior: // - Accepts 1..10,000 rows in one gRPC request @@ -37,9 +37,27 @@ service TablesData { // - Each row is inserted through the same validation, script execution, // typed binding, database insert, and indexing path as PostTableData // - Stops at the first failing row and returns that row's gRPC error code - // with row index context; rows inserted before the failure remain inserted + // - Commits only when every row succeeds; a failure rolls the entire request back rpc PostTableDataBulk(PostTableDataBulkRequest) returns (PostTableDataBulkResponse); + // Starts a durable, profile-scoped import staging session. Staging never changes + // profile data; CommitTableDataImport applies every staged chunk atomically. + rpc BeginTableDataImport(BeginTableDataImportRequest) + returns (BeginTableDataImportResponse); + + // Adds one table chunk to an import session. Rows retain chunk and table order. + rpc StageTableDataImport(StageTableDataImportRequest) + returns (StageTableDataImportResponse); + + // Applies every staged row through the ordinary validated insert machinery in + // one PostgreSQL transaction. Any failure rolls back every imported side effect. + rpc CommitTableDataImport(CommitTableDataImportRequest) + returns (CommitTableDataImportResponse); + + // Discards a staged import. Already committed imports cannot be aborted. + rpc AbortTableDataImport(AbortTableDataImportRequest) + returns (AbortTableDataImportResponse); + // Update existing row data with strict type binding and script validation. // // Behavior: @@ -201,6 +219,42 @@ message PostTableDataBulkResponse { repeated PostTableDataResponse responses = 3; } +message BeginTableDataImportRequest { + string profile_name = 1; +} + +message BeginTableDataImportResponse { + string import_id = 1; +} + +message StageTableDataImportRequest { + string import_id = 1; + string table_name = 2; + repeated PostTableDataBulkRow rows = 3; +} + +message StageTableDataImportResponse { + int64 staged_rows = 1; + int64 total_staged_rows = 2; +} + +message CommitTableDataImportRequest { + string import_id = 1; +} + +message CommitTableDataImportResponse { + bool success = 1; + int64 inserted_rows = 2; +} + +message AbortTableDataImportRequest { + string import_id = 1; +} + +message AbortTableDataImportResponse { + bool success = 1; +} + // Update an existing row. message PutTableDataRequest { // Required. Profile (schema) name. diff --git a/common/src/proto/descriptor.bin b/common/src/proto/descriptor.bin index ddd598dbc09369f9840e7303cb329da8a1f9b838..8035716c24a0b94e0c47b774abcad2a75e1af12f 100644 GIT binary patch delta 9090 zcmb7}dvsORoyYI~?UR$66Ox;pyx`sh4zHp}0vJ&qRm4Y;B^?y3tWFtO*W^MngoNf^ z>|)W*wTQUDTC@tMzEEjV5wKxUVps@@3_%%?mqT?_rnqY9DzZukt>B~c{q4uSvs{^3 zGymmt_x}FQ@Auu$bMna6;tjpUcP|$&(Svu>FZOl+;%?gO)LiQ&u1n2tZNDvfXItuq zWF|Rn;i8Vt%)>`!kIW94xKPAmpC^|9Jd(ueNWkaw^>a6Oo7x z*+O2$EsHbZ9rx3$?yh_3KTBvwtb4=5loA7$dX;^Rxj!a+=pL%-vn~9OpNg8WY%LXD z*_SjoT@YH?{EfP7=)T(Wp~v;roD2O=UR9s2uU$(u1N&?-gBibE)pgbX(8tU#WZsvn zHuJvP@|pK3?G|gO#qplLUwDOnfruOrJ0GQH@#F3jkJ14r_L1})H$vpgOX&9aJK@%K zRO2Nx9Sd8Vl5K5Ex_j4QtY*{)I_-4O(cZQs_)$x$JxJY^>Rb|F83gIYO--p(b9z#c z4CW@x7f>e8^m4DEyF}O3?v>?3eX&RN?B-Ep&?~J1Q`uGzffb*T1 zG@>*w?&NJ^RIsSCV_s`psy^M+u_)CX=rtdtu>@P&=LhLjI^EjQ-oQfjLpvuGG_@q# z=cm%1RR_)3ZYG%$a&sc9Cwq`ww5Y8$l{WQoXgsK?WpVq0An!wi=k@anV3ZWT9OcF> z!M9O~D=kT1*2gaM{z28{_|6y_TOXehOlfXr<5EF}i!!k@*^VIDo~y`K!s`-VWs*(CRbVkN*m`UjI9MIZCvemODOHi z*eLIIWwPr#6KG<6{HEZB)^t;{6GkU(QF^j81XJ2NQ_1EfL6h2RRV{p$Ovg-XO18Im zWP&?WL6Y~uRC6u`7b#50x$7c@2|0INq%aZg{r5>)9)v%Sq)Rp!g3-7mjDBSk|QR7-OsSO(S7+z6IIwe-gCdx1wz z#4pH3o()SjQxMBWYrP+OjuWAX?<^DHq|MYap~NfTAFyPX35t3K1!GZ_MeLc&ME?Qa zy%-YoM(YX@{&q9{c)ac}AYYM}3)P{h!pH^WD@1jT_mHRLS8kz-g6swn8De^I z)D2?j2#XMH5F;;x6L86QpBIr$LU_fVbU6I;BFYopaRe8$QP3cz17V2*t1Y7MMo|;D zh^QMy-9RI~7OgJ`y4JJ`Xo0xPz!X9JoEOBfNfx1fL0mIg`AWX~qKLd2{$LvoioGbJ zbzVJ^Ax@?%T)B-#*LMkWl@dQ9P?snywTM7nqTKgJCxLsP>+SAlc5|k5>{@X*V2bHg4%!;}&~BG4W&{8Ebuv`37p_l6;mQpV{GjJDnh|opoy$Ok+|P3vTN1=* zJP`hMJI$y+ke9IS2ZWiw=mW_C_VDkO1hyXv)7>=Z>OdaBkLJjY&?V&re@7WRxR0;2i;y`&1M1c3i z;E@&);5|{#`z<-H?|vX6C(U}q+2wp7qC?f9a=DHVF&;_>LRvsQ8bI{@(2j;nSuW^9 z9u5v$=)1oYk&jGR9B$6FTZRZFOkxtt=0+fChj7h>t7y0g=MC7bV8b_@2Cu>C!=X}N{ z2ttBDy#WBx_l(^DfJpF+O|Xcc`H7%`sx}a$Kr791ai56j0NV`hC!)4aW#swppGD*^ zCLSwRY%P_IO3i2BFgs^$0_Ep zp9<1@83u3A3uue!hD_ zM7}VE;`-oRu=AxKXLQjP3PMIeT__NJFWN$Z2z${MTH?E3iOAnfR9r8dudIC~9QA9| zIZ^ozL|Cvc6o}4W+d_c|`?cN!$SL*RWfXaW!Y6uYa3yXa&N7N7yfLt^l*2Bk@JJ6` zb_D=Imy^jAh_1`Y9PvN|T~1hsd`^_|G`Ww$F*~RcN}eV6<&|vSePmVuTA{p;umW%$ zf)XoW6@}prsvie{aupfdfl#hO2C7UTl&h$u)H~!TDKMn&3p;2UHI|aZY6;1l3_xf@ zGA9F2b;ijsOG&ZN9t%r*>DDZtOpoOs$tTleWM*_JpGH74qhr3ijv~JzeXGDd%vncf z>c!Z*r-%ltR6$4vs3!~%eV?N0THA`KPtl+N(BKTJ(GV9`v=q$0gaG8LxuX(1w#4^VNcgm(0aTSj5SF~c(R_%<`W}{Q0u9< z+_oardh$nERN~Qm?WXT?6cY! z$0_6VpCgkX2nho9Yycv`=WK#N)!8HnM5N~k2?j{8+;=xoWJmboZc0>^qvIyRwc1~DsLZJi`HFaPEcQY3A9@F za+lpd&*9}D2Uc2K05w_P_R>_s^x0Zn*1zIUvw%=NYR!H}2%dp>6@}}{o z^$Y|pPVHwa;YdK&|x?;*Pkfe3qummzyt#p%9HRHsG?1SwEw1XQh2eK!M2`2RK~ z_)7+BVZ>n~eSHBz3)Ch8p*>7R`ZWiL#~$VtrKO-Fc7jw9Z=OIsL4b&K#7+<(A|25a zBwK*TvN1>W4B;%GJxZhx2N1MCTLG`fqZHNmShPZVl;ZsDiKXa!EMGu@Mk~c2=N$z$=pos?*W@_n9^eD|p4p*R`-W*?1$kjpx0EdWAzlA`)3 z0z!C_Dy!5{Ud?ohkMbE;;BN>|<%R6~DLWt0%3b*|=eMdF-~E^(e+(b$rJ$k)9Y5w} zc^z`8;rdU9dwc1c@%S4dPusZ&M9tqG|$Cx2-)9v&rswu67dn?SI^S$aPEF8 ztn3fP8GFa-&#})MZvhB?);a`4*R$4JAcCK@-u7n~&KYk3P@c;x+1qp0TeL!Xj=dd= zdit}s=flZwQhgSGLwMd=1cdOsUF|>!&)W@hfbV`zk*~~(9e|#nQ*?-$c>}nP3wG-Q zAu3QW9U%H%u!R5-^#T{7UYG{(GcVe`3j{6DYPpn)wiL8Nd(m$91Gtng?A8T>6sRXW z5YjJfF+fPauv^yvrpaFt>GceR7HG9RSH7gEK1k4t-d~bm;q4P16-0b@nT)KG;l{UU zZ+4k1P#Z{r`Y?8nB=w<&2{C|0ycaPd*?VNa5_SxT#^sXKEN0T!&-4}OyIjV~OkdEw zlGG&teOU@D(S5HYcHv%GR&F{Y)fM?9xDcRC0#SfS;t_m{Lo94ID`nX5ue2t+Qj&c< zKF{N4^>}jPzWXy7c}VJeZXEG`Ce1M)=W%~P8WTY<5vWZBqVEIJm4@j8En^l~v z__;J6-auL01;P3*2v*J2`a1v++Mi4Fr;X$l%y*MAr#PtVw=KzD7Gn-zWx-1 z&PqF!zr>JJ>${K1$Wz9qTJ(HO;{8g!`qZ*dk4tlX10gI>`vgSi$EA7y10w9>Qr+Gh zrNiFJMm;*3tA`5 zYi6x_>jdg61Q617GOj=C0afQAL%pkEM4j)hmyz94#777CH(C&W`!J2Js)J&^#Oug8 ze6x4z*pN-K`|iWkCdPq~*(N*qKy=!YXpc$U3Tp2*q^OxjbMp@5DBQq9*9U=^O5+eTkR4-D{t*$ndC0}}rpKl^X=oytMzctGOg17|acvpHz80U+i< z8xx4G2W>V$#5`!T8N}Hf*4Z=ypgf#cayEx;HfV+NFlX}v^-Ro>V`0f}Y4(^wYDyfl bqXUHSn2hT04nPQx$%@K5UHrQ|)6w`}`N&*( delta 6586 zcmYM3dyG}py~oe`?S1x~+4J6K9yq)X0zQzL0cTK)MWl_r5o5j9216gW!OIY*msfC% zCikWXB2jFVI)HB83BgBIv{NOF%0mha<+hZHaBI=1%)R)k^olnjCJ@s5`K`4V^WS{V z`h9=vx4!GQ*M7`%zpq;N+p6U&%}#l0xvc0PShHLXg#+gvlRInWVAa6PXQbNx z3%g6pmOS*p!2Inbnq4YA)YE&v{!;2$Jiq(l(!VZQ+}r(d@69%d-&EN>zjR;sf+fqk zOTG7Zmln+HS@FEE86!JAp_y^Zz40bvMHNXf zD?R_RZ0M*!G%IS(!YUBWYH9a`Xx6vxj?an}6_Q}~z}_A5bJI~kH9Km`;tHr{H&6D2 zYW9uakN@E|uQCbd49s~|E}8a9RC6vHScz)RWdkcw&B4Iy?No6R%uQGH%ao27&D`jU zEO(4%Zfln(G;?R(6F=(et4e}-1E>2XG#yo_=0z=8TotN$Bc^ylHSgwo;~h>_odi7t z#n)t#X|G1rb9tkxQT1HjsA^O_x6O@9v4puKzbLmZ{kPYpRJ*7!Dqb3gVNP;Mur%HH zy4*jF22o}=mr6lrRU-G&O6!;s*_MvDB3{XMfQ}e$XKg`B8$x)iA_yQt4Fe?X7rd$dMl&rjog}z8nobek9Ksz%^uONH@D2R zeo7K;Ho2G6-@PSc>o%Lh=(vqyM2p#)e)X1gv~4vKStUP`XsfBH^+=+vrXh*1wh{)n zrS>G?e}rS4Tj>)lS?K#l58yY|W*?fph3SS3G_ zsNYPy)+34f&E&Rt0*S&T+G%oo(^vM&_`02@P>Sy&QK*S_r62E=yW4ho=lGFCy9~Xn zMuH^TWoTZ`ACSQ})1U5>yV~9ymZ;fp8hX}A=vffweq<$>y(c}jPku6EkGFvhqF?ry zD%X0tUy3ES7?2Em%+=S#6RAF*MEgzdKhtGz%Y?f9rZ6u4HhJ?}{ekqkx8;_$1HNDQ zkwgc4zknn<;QJ-7vkYp#T-!G2tLI1k)wc?WWKdh>Cw3O9KawukFaL7O5np}YR)54) zmF$8+G8{4ECV3>o5!2RAlS&vR(NUB8)TFN*l5ve3+Tl@C7#U9?d8A7?X3~rA$aP%+ zopH>U1rmD9?*mBEW4aI8j*-rI+@!MxrE~35H%0BohjqG~<9;&;M0Z>_gQFH5%Qt`F=Ul4TZAH4n%pz6t1x5fhftY4V1rMT?0jQ-6ZUKOH;7w8Tz(2iz@7l`(6dU!c0 z4t_a2k6H-2Kwa{x9=|+jIDpDNL}l6x5e<-WQ0qfzFqLEK+06d1DwR?Kw3@D=cUS zeC-Q_C=hf7f)x0*FA%iBS%Dy#zSaWuo>`eh=S^;9`t~1WWFrp(;dwp2zQ-JuTHiM& z{og;xtWJQO-+2E(Lcj5sFG$XB{N-Dz)m^a1eHVWtyx_YY)G>b%g9tCE@ssS7?Aaxg zzWGO)$dtNemwf+$C@+~pVtoctUNTM1@r_mqEl(ScN+&|ypyiS^E{JfsWQRYfp~B&> zhjgVr6h4wZbX0CD^S6RzK7!^x@_ z=_=s}aZ=2DT%<0Os+*v@iN@B=$4TTzgQUzbhlsAGDv6$u-0EyKJhg>S2*pg?eg3dD# z?ON{{h<2^BOr_8|kuh0K+Jm5%Y7efH?C>Pk24{z7O%kP&+gi??s3A1<%~PXBtQT>& zY>2d=JH|j_*Gu-)1(J5X@YJOq)@ZpKL`K_+Af%wH2(+UG*I8hI{~Khaz7MGuBQ}b- z!yZBlIuk*(8>P}cPJv9jQLiW`g*N#HsS!OqLDwK4nKtE?Ff=>ySAe~0o1Bj&!@Kz z@<96huteQ|-nSt!NS@a=)W?llHUGtQ|8bez$=_7|qF*$K@I@)On-7TaMQLoZ_wQPz z9qHJ2<*qV+Biu18RHt|N4rEkAg$~pwx4IUF9&)b`1p{7uR?yptf2Z+P=T zQoiBMuUEHrtNC3B^#a&EEYvO7?YDqYgu8VMrn0zt-GY7TKcAGgGJhl7=bZ!*?(@e0 zh;W}jIUADbfaHFkUCIqa9+1M7wj~>M6$ky929XtX*Az(XLB9@=tOs=+_CcsY=RD*O zH3%)}wYro;ekqKiJ>>5L4Z4&;f2Ki5L3bs9NC*95K%|5IOlwe@d050<=pb6qYqg~g zOTpb?7$y3!B*pl!i6x)od0KKOC4K9^<1poy8hrk4as977P8EwPu4V1#29BN7RKsOMOEYOsVe-7oiIQeGz;^ z0x4o0-vOAHQ|N@GGvAY^%O^zq^dV6v-1I~w(O)F@LALJ+`TpXM@I+Vqo;MM~M9`TC z68oMv5u~#B)I>eS5?#e#C42FQ$~+{(?#2h(;b!+_0iyk@WN-e7uK1LQd(nhg1?bLU zkSb33a~LGkDM{RW9Z05A!h4;o0(xI$Qe0E!ew)}M9rmfTE8ZVYrDnb_$wYNtgRS_Q zYe^+xF^N8u+<#{_6^Z=N-|~xU(?{MWh^(Nq2_*I-DQC?FllCL;Q&G$R*!u)w6XTanL=q&n?=VwhB?y-!7@ArZH7h*Y3Yr8W%> zr&61SByq2%1Y7NReK(M>F^SGd?u*Q(Mk3GX$z2`ujMAt+ed4cYh_s;d2_*Ove?5bw z{ls6-javPuzCZ|@K-WH?9UbW^0+D_ywj!Msn?7UxZgC*m&;0!rMEjYvdaXL~*{~M+ zT(U19jrNTcbk7zb($A&UePIA?@Rj4^85q%&L}w-UcwkyPN(#$(VkXw@mLvm$fGl5S8yb$U0O|x#;>OeoIOl@BsWOp%sC1`b!J=KF` zTODMNQO#;pUtm9^bOF@SH>}kC=nJy4NT{j+P{PI^aJT{}__*HMS7=bikJh zt)oTjNNpY0bphl|eNK?j)K>$NGxgQ9Xf+#MHJt#;jl)W5eVw mnE#P`r?lA9WsBbz5aE`f;66%#2)6{q#`~hB%a-&m>HL4Y4-)18 diff --git a/common/src/proto/komp_ac.tables_data.rs b/common/src/proto/komp_ac.tables_data.rs index 2fd878ab..9be0e14b 100644 --- a/common/src/proto/komp_ac.tables_data.rs +++ b/common/src/proto/komp_ac.tables_data.rs @@ -113,6 +113,54 @@ pub struct PostTableDataBulkResponse { #[prost(message, repeated, tag = "3")] pub responses: ::prost::alloc::vec::Vec, } +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct BeginTableDataImportRequest { + #[prost(string, tag = "1")] + pub profile_name: ::prost::alloc::string::String, +} +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct BeginTableDataImportResponse { + #[prost(string, tag = "1")] + pub import_id: ::prost::alloc::string::String, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct StageTableDataImportRequest { + #[prost(string, tag = "1")] + pub import_id: ::prost::alloc::string::String, + #[prost(string, tag = "2")] + pub table_name: ::prost::alloc::string::String, + #[prost(message, repeated, tag = "3")] + pub rows: ::prost::alloc::vec::Vec, +} +#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] +pub struct StageTableDataImportResponse { + #[prost(int64, tag = "1")] + pub staged_rows: i64, + #[prost(int64, tag = "2")] + pub total_staged_rows: i64, +} +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct CommitTableDataImportRequest { + #[prost(string, tag = "1")] + pub import_id: ::prost::alloc::string::String, +} +#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] +pub struct CommitTableDataImportResponse { + #[prost(bool, tag = "1")] + pub success: bool, + #[prost(int64, tag = "2")] + pub inserted_rows: i64, +} +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct AbortTableDataImportRequest { + #[prost(string, tag = "1")] + pub import_id: ::prost::alloc::string::String, +} +#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] +pub struct AbortTableDataImportResponse { + #[prost(bool, tag = "1")] + pub success: bool, +} /// Update an existing row. #[derive(Clone, PartialEq, ::prost::Message)] pub struct PutTableDataRequest { @@ -583,7 +631,7 @@ pub mod tables_data_client { ); self.inner.unary(req, path, codec).await } - /// Insert multiple rows by applying PostTableData behavior to each row. + /// Insert multiple rows atomically by applying PostTableData behavior to each row. /// /// Behavior: /// @@ -592,7 +640,7 @@ pub mod tables_data_client { /// * Each row is inserted through the same validation, script execution, /// typed binding, database insert, and indexing path as PostTableData /// * Stops at the first failing row and returns that row's gRPC error code - /// with row index context; rows inserted before the failure remain inserted + /// * Commits only when every row succeeds; a failure rolls the entire request back pub async fn post_table_data_bulk( &mut self, request: impl tonic::IntoRequest, @@ -622,6 +670,128 @@ pub mod tables_data_client { ); self.inner.unary(req, path, codec).await } + /// Starts a durable, profile-scoped import staging session. Staging never changes + /// profile data; CommitTableDataImport applies every staged chunk atomically. + pub async fn begin_table_data_import( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { + self.inner + .ready() + .await + .map_err(|e| { + tonic::Status::unknown( + format!("Service was not ready: {}", e.into()), + ) + })?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static( + "/komp_ac.tables_data.TablesData/BeginTableDataImport", + ); + let mut req = request.into_request(); + req.extensions_mut() + .insert( + GrpcMethod::new( + "komp_ac.tables_data.TablesData", + "BeginTableDataImport", + ), + ); + self.inner.unary(req, path, codec).await + } + /// Adds one table chunk to an import session. Rows retain chunk and table order. + pub async fn stage_table_data_import( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { + self.inner + .ready() + .await + .map_err(|e| { + tonic::Status::unknown( + format!("Service was not ready: {}", e.into()), + ) + })?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static( + "/komp_ac.tables_data.TablesData/StageTableDataImport", + ); + let mut req = request.into_request(); + req.extensions_mut() + .insert( + GrpcMethod::new( + "komp_ac.tables_data.TablesData", + "StageTableDataImport", + ), + ); + self.inner.unary(req, path, codec).await + } + /// Applies every staged row through the ordinary validated insert machinery in + /// one PostgreSQL transaction. Any failure rolls back every imported side effect. + pub async fn commit_table_data_import( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { + self.inner + .ready() + .await + .map_err(|e| { + tonic::Status::unknown( + format!("Service was not ready: {}", e.into()), + ) + })?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static( + "/komp_ac.tables_data.TablesData/CommitTableDataImport", + ); + let mut req = request.into_request(); + req.extensions_mut() + .insert( + GrpcMethod::new( + "komp_ac.tables_data.TablesData", + "CommitTableDataImport", + ), + ); + self.inner.unary(req, path, codec).await + } + /// Discards a staged import. Already committed imports cannot be aborted. + pub async fn abort_table_data_import( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { + self.inner + .ready() + .await + .map_err(|e| { + tonic::Status::unknown( + format!("Service was not ready: {}", e.into()), + ) + })?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static( + "/komp_ac.tables_data.TablesData/AbortTableDataImport", + ); + let mut req = request.into_request(); + req.extensions_mut() + .insert( + GrpcMethod::new( + "komp_ac.tables_data.TablesData", + "AbortTableDataImport", + ), + ); + self.inner.unary(req, path, codec).await + } /// Update existing row data with strict type binding and script validation. /// /// Behavior: @@ -992,7 +1162,7 @@ pub mod tables_data_server { tonic::Response, tonic::Status, >; - /// Insert multiple rows by applying PostTableData behavior to each row. + /// Insert multiple rows atomically by applying PostTableData behavior to each row. /// /// Behavior: /// @@ -1001,7 +1171,7 @@ pub mod tables_data_server { /// * Each row is inserted through the same validation, script execution, /// typed binding, database insert, and indexing path as PostTableData /// * Stops at the first failing row and returns that row's gRPC error code - /// with row index context; rows inserted before the failure remain inserted + /// * Commits only when every row succeeds; a failure rolls the entire request back async fn post_table_data_bulk( &self, request: tonic::Request, @@ -1009,6 +1179,40 @@ pub mod tables_data_server { tonic::Response, tonic::Status, >; + /// Starts a durable, profile-scoped import staging session. Staging never changes + /// profile data; CommitTableDataImport applies every staged chunk atomically. + async fn begin_table_data_import( + &self, + request: tonic::Request, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; + /// Adds one table chunk to an import session. Rows retain chunk and table order. + async fn stage_table_data_import( + &self, + request: tonic::Request, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; + /// Applies every staged row through the ordinary validated insert machinery in + /// one PostgreSQL transaction. Any failure rolls back every imported side effect. + async fn commit_table_data_import( + &self, + request: tonic::Request, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; + /// Discards a staged import. Already committed imports cannot be aborted. + async fn abort_table_data_import( + &self, + request: tonic::Request, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; /// Update existing row data with strict type binding and script validation. /// /// Behavior: @@ -1353,6 +1557,190 @@ pub mod tables_data_server { }; Box::pin(fut) } + "/komp_ac.tables_data.TablesData/BeginTableDataImport" => { + #[allow(non_camel_case_types)] + struct BeginTableDataImportSvc(pub Arc); + impl< + T: TablesData, + > tonic::server::UnaryService + for BeginTableDataImportSvc { + type Response = super::BeginTableDataImportResponse; + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { + ::begin_table_data_import(&inner, request) + .await + }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = BeginTableDataImportSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } + "/komp_ac.tables_data.TablesData/StageTableDataImport" => { + #[allow(non_camel_case_types)] + struct StageTableDataImportSvc(pub Arc); + impl< + T: TablesData, + > tonic::server::UnaryService + for StageTableDataImportSvc { + type Response = super::StageTableDataImportResponse; + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { + ::stage_table_data_import(&inner, request) + .await + }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = StageTableDataImportSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } + "/komp_ac.tables_data.TablesData/CommitTableDataImport" => { + #[allow(non_camel_case_types)] + struct CommitTableDataImportSvc(pub Arc); + impl< + T: TablesData, + > tonic::server::UnaryService + for CommitTableDataImportSvc { + type Response = super::CommitTableDataImportResponse; + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { + ::commit_table_data_import(&inner, request) + .await + }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = CommitTableDataImportSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } + "/komp_ac.tables_data.TablesData/AbortTableDataImport" => { + #[allow(non_camel_case_types)] + struct AbortTableDataImportSvc(pub Arc); + impl< + T: TablesData, + > tonic::server::UnaryService + for AbortTableDataImportSvc { + type Response = super::AbortTableDataImportResponse; + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { + ::abort_table_data_import(&inner, request) + .await + }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = AbortTableDataImportSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } "/komp_ac.tables_data.TablesData/PutTableData" => { #[allow(non_camel_case_types)] struct PutTableDataSvc(pub Arc); diff --git a/web/locales/cs/main.ftl b/web/locales/cs/main.ftl index bfb6e40d..8d363df2 100644 --- a/web/locales/cs/main.ftl +++ b/web/locales/cs/main.ftl @@ -592,9 +592,6 @@ import-continue = Pokračovat import-carried = Import do tabulky { $table } v rozsahu { $scope }. import-error-title = CSV se nepodařilo importovat import-import-rows = Importovat -import-failure-message = Před zastavením importu se importovalo { $inserted } z { $source_rows } řádků. Selhal řádek CSV { $row }. - Backend: { $error } - Již importované řádky zůstávají v tabulce. Aby nevznikly duplicity, neimportujte znovu celý soubor: nejprve tyto řádky odstraňte nebo importujte pouze chybný a zbývající řádky. import-failure-no-rows-message = Nic se neimportovalo. Selhal řádek CSV { $row }. Backend: { $error } import-err-bad-date = Řádek CSV { $row }, sloupec { $column }: „{ $value }“ není platné datum ve formátu { $format }. @@ -611,7 +608,7 @@ import-success-message = Vloženo { $inserted -> # --- Import v průběhu ------------------------------------------------------- import-progress-heading = Importuje se do { $table } -import-progress-rows = Importováno { $inserted } z { $total } řádků +import-progress-rows = Připraveno { $inserted } z { $total } řádků import-progress-label = Průběh importu import-progress-elapsed = Uplynulo import-progress-rate = Rychlost diff --git a/web/locales/en/main.ftl b/web/locales/en/main.ftl index 06b5e8f5..3a6407cd 100644 --- a/web/locales/en/main.ftl +++ b/web/locales/en/main.ftl @@ -582,9 +582,6 @@ import-continue = Continue import-carried = Importing into { $table } in { $scope }. import-error-title = Could not import CSV import-import-rows = Import -import-failure-message = Imported { $inserted } of { $source_rows } rows before the import stopped. CSV row { $row } failed. - Backend: { $error } - Rows already imported remain in the table. To avoid duplicates, do not retry the whole file: remove those rows first, or import only the failed and remaining rows. import-failure-no-rows-message = Nothing was imported. CSV row { $row } failed. Backend: { $error } import-err-bad-date = CSV row { $row }, column { $column }: “{ $value }” is not a valid { $format } date. @@ -599,7 +596,7 @@ import-success-message = Inserted { $inserted -> # --- The import while it runs ---------------------------------------------- import-progress-heading = Importing into { $table } -import-progress-rows = { $inserted } of { $total } rows imported +import-progress-rows = { $inserted } of { $total } rows staged import-progress-label = Import progress import-progress-elapsed = Elapsed import-progress-rate = Speed diff --git a/web/locales/sk/main.ftl b/web/locales/sk/main.ftl index 09d69b82..bc81bafb 100644 --- a/web/locales/sk/main.ftl +++ b/web/locales/sk/main.ftl @@ -592,9 +592,6 @@ import-continue = Pokračovať import-carried = Import do tabuľky { $table } v rozsahu { $scope }. import-error-title = CSV sa nepodarilo importovať import-import-rows = Importovať -import-failure-message = Pred zastavením importu sa importovalo { $inserted } z { $source_rows } riadkov. Zlyhal riadok CSV { $row }. - Backend: { $error } - Už importované riadky zostávajú v tabuľke. Aby nevznikli duplicity, neimportujte znova celý súbor: najprv tieto riadky odstráňte alebo importujte iba chybný a zostávajúce riadky. import-failure-no-rows-message = Nič sa neimportovalo. Zlyhal riadok CSV { $row }. Backend: { $error } import-err-bad-date = Riadok CSV { $row }, stĺpec { $column }: „{ $value }“ nie je platný dátum vo formáte { $format }. @@ -609,7 +606,7 @@ import-success-message = Vložený { $inserted -> # --- Import počas behu ------------------------------------------------------ import-progress-heading = Importuje sa do { $table } -import-progress-rows = Importovaných { $inserted } z { $total } riadkov +import-progress-rows = Pripravených { $inserted } z { $total } riadkov import-progress-label = Priebeh importu import-progress-elapsed = Uplynulo import-progress-rate = Rýchlosť diff --git a/web/src/pages/import_export/import/logic.rs b/web/src/pages/import_export/import/logic.rs index 064b65ec..944bb603 100644 --- a/web/src/pages/import_export/import/logic.rs +++ b/web/src/pages/import_export/import/logic.rs @@ -13,7 +13,10 @@ use crate::{ AppState, definitions::{ table_structure::GetTableStructureRequest, - tables_data::{PostTableDataBulkRequest, PostTableDataBulkRow}, + tables_data::{ + AbortTableDataImportRequest, BeginTableDataImportRequest, CommitTableDataImportRequest, + PostTableDataBulkRow, StageTableDataImportRequest, + }, }, services::{authenticated_request, reject_cross_site}, {i18n::Locale, tr}, @@ -264,7 +267,10 @@ pub(crate) async fn import_csv( // practice; it is not worth a panic on the import path. None => ( StatusCode::INTERNAL_SERVER_ERROR, - Html(ui::render_error(locale, &tr!(locale, "import-progress-gone"))), + Html(ui::render_error( + locale, + &tr!(locale, "import-progress-gone"), + )), ) .into_response(), } @@ -350,37 +356,100 @@ struct Running { /// The insert, chunk by chunk, reporting after each one. async fn run_import(job: Running) { let jobs = job.state.imports.clone(); - let mut inserted = 0usize; + let begin = BeginTableDataImportRequest { + profile_name: job.profile_name.clone(), + }; + let Ok(begin) = authenticated_request(&job.headers, begin) else { + return jobs.finish(&job.id, 0, Outcome::SessionLost); + }; + let mut data = job.state.tables_data.clone(); + let import_id = match data.begin_table_data_import(begin).await { + Ok(response) => response.into_inner().import_id, + Err(error) => { + let (_, outcome) = import_failure(&error, 0, 0); + return jobs.finish(&job.id, 0, outcome); + } + }; + let mut staged = 0usize; - for (chunk_index, chunk) in job.rows.chunks(CHUNK_ROWS).enumerate() { - let request = PostTableDataBulkRequest { - profile_name: job.profile_name.clone(), + for chunk in job.rows.chunks(CHUNK_ROWS) { + let request = StageTableDataImportRequest { + import_id: import_id.clone(), table_name: job.table_name.clone(), rows: chunk.to_vec(), }; let Ok(request) = authenticated_request(&job.headers, request) else { - return jobs.finish(&job.id, inserted, Outcome::SessionLost); + abort_import(&job, &import_id).await; + return jobs.finish(&job.id, 0, Outcome::SessionLost); }; - let mut data = job.state.tables_data.clone(); - match data.post_table_data_bulk(request).await { + match data.stage_table_data_import(request).await { Ok(response) => { - inserted += response - .into_inner() - .responses - .iter() - .filter(|row| row.inserted_id > 0) - .count(); - jobs.advance(&job.id, inserted); + staged = usize::try_from(response.into_inner().total_staged_rows) + .unwrap_or(job.rows.len()); + jobs.advance(&job.id, staged); } Err(error) => { - let (inserted, outcome) = - import_failure(&error, inserted, chunk_index * CHUNK_ROWS); - return jobs.finish(&job.id, inserted, outcome); + abort_import(&job, &import_id).await; + let (_, outcome) = import_failure(&error, 0, 0); + return jobs.finish(&job.id, 0, outcome); } } } - jobs.finish(&job.id, inserted, Outcome::Succeeded); + let mut commit_attempt = 0; + let committed = loop { + let Ok(commit) = authenticated_request( + &job.headers, + CommitTableDataImportRequest { + import_id: import_id.clone(), + }, + ) else { + abort_import(&job, &import_id).await; + return jobs.finish(&job.id, 0, Outcome::SessionLost); + }; + match data.commit_table_data_import(commit).await { + Err(error) + if commit_attempt == 0 + && matches!( + error.code(), + tonic::Code::Cancelled + | tonic::Code::Unknown + | tonic::Code::DeadlineExceeded + | tonic::Code::Internal + | tonic::Code::Unavailable + ) => + { + // The first commit may have reached PostgreSQL even if its response was lost. + // Completed sessions are durable, so repeating this call cannot duplicate rows. + commit_attempt += 1; + } + result => break result, + } + }; + match committed { + Ok(response) => { + let inserted = usize::try_from(response.into_inner().inserted_rows).unwrap_or(staged); + jobs.finish(&job.id, inserted, Outcome::Succeeded); + } + Err(error) => { + abort_import(&job, &import_id).await; + let (_, outcome) = import_failure(&error, 0, 0); + jobs.finish(&job.id, 0, outcome); + } + } +} + +async fn abort_import(job: &Running, import_id: &str) { + let Ok(request) = authenticated_request( + &job.headers, + AbortTableDataImportRequest { + import_id: import_id.to_string(), + }, + ) else { + return; + }; + let mut data = job.state.tables_data.clone(); + let _ = data.abort_table_data_import(request).await; } /// POST /admin/import/prepared.csv — the same file the import would read, to @@ -487,8 +556,7 @@ async fn prepared( ) -> Result<(Destination, Source, Prepared), Response> { let locale = Locale::from_headers(headers); let destination = destination(state, headers, form).await?; - let source = read_source(locale, &form.csv_data) - .map_err(|message| reject(headers, message))?; + let source = read_source(locale, &form.csv_data).map_err(|message| reject(headers, message))?; // Every posted destination has to still exist and still be writable. A key // that resolves to nothing is refused rather than skipped: skipping it @@ -511,13 +579,8 @@ async fn prepared( let assignments = read_mapping(locale, &chosen, &source).map_err(|message| reject(headers, message))?; let mut prepared = prepare(&assignments, &source, &destination.names()); - normalize_dates( - locale, - &mut prepared, - &destination.types, - form.date_format, - ) - .map_err(|message| reject(headers, message))?; + normalize_dates(locale, &mut prepared, &destination.types, form.date_format) + .map_err(|message| reject(headers, message))?; Ok((destination, source, prepared)) } @@ -604,24 +667,21 @@ fn grpc_error(headers: &HeaderMap, error: &tonic::Status) -> Response { .into_response(Locale::from_headers(headers), ui::render_error) } -/// How a failed chunk ends the import: the final inserted count, and what the -/// page will say about it. -/// -/// The rows before the failure are in the table and stay there, so the count -/// the job keeps is the count the message is built from. +/// How a failed atomic commit ends the import. Staging progress is discarded: +/// a failed import commits no profile rows. fn import_failure( error: &tonic::Status, - inserted_before_chunk: usize, + _inserted_before_chunk: usize, chunk_start: usize, ) -> (usize, Outcome) { let failure = crate::ui::FormError::from_status_with_message(error, format!("{error:?}")); if matches!(failure, crate::ui::FormError::Unauthenticated) { - return (inserted_before_chunk, Outcome::SessionLost); + return (0, Outcome::SessionLost); } let status = failure.status_code(); match bulk_failure(error) { - Some((failed_row_index, inserted_in_chunk)) => ( - inserted_before_chunk + inserted_in_chunk, + Some((failed_row_index, _inserted_in_chunk)) => ( + 0, Outcome::RowFailed { status, // One header row precedes the data, and CSV rows are one-based. @@ -629,10 +689,9 @@ fn import_failure( backend: format!("{error:?}"), }, ), - // The backend refused the batch without saying which row did it, so - // there is no partial progress to report inside this chunk. + // The backend refused the import without identifying one row. None => ( - inserted_before_chunk, + 0, Outcome::Refused { status, message: failure.message().to_string(), diff --git a/web/src/pages/import_export/import/progress.rs b/web/src/pages/import_export/import/progress.rs index b6f88656..e2bba2a7 100644 --- a/web/src/pages/import_export/import/progress.rs +++ b/web/src/pages/import_export/import/progress.rs @@ -33,10 +33,9 @@ const KEEP_FINISHED: Duration = Duration::from_secs(300); /// in their language, in whichever alert their page expects. #[derive(Clone, Debug)] pub(crate) enum Outcome { - /// Every chunk was accepted. + /// Every staged row was committed atomically. Succeeded, - /// The backend stopped on one row. `inserted` on the snapshot says how much - /// of the file went in before it, and stays true: those rows remain. + /// The backend stopped on one row. No rows from the import were committed. RowFailed { status: StatusCode, /// One-based row of the file the user handed over, header included. @@ -222,8 +221,7 @@ mod tests { assert!(snapshot.outcome.is_none()); } - /// A failure keeps the count it reached: those rows are in the table, and - /// the message about what to do next is built from that number. + /// Finishing replaces staging progress with the number actually committed. #[test] fn a_finished_job_keeps_its_count_and_its_outcome() { let jobs = ImportJobs::default(); @@ -232,7 +230,7 @@ mod tests { jobs.advance(&id, 300); jobs.finish( &id, - 340, + 0, Outcome::RowFailed { status: StatusCode::UNPROCESSABLE_ENTITY, csv_row: 342, @@ -241,7 +239,7 @@ mod tests { ); let snapshot = jobs.snapshot(&id, "session").unwrap(); - assert_eq!(snapshot.inserted, 340); + assert_eq!(snapshot.inserted, 0); assert!(matches!( snapshot.outcome, Some(Outcome::RowFailed { csv_row: 342, .. }) diff --git a/web/src/pages/import_export/import/ui.rs b/web/src/pages/import_export/import/ui.rs index eb03a372..559c9f5c 100644 --- a/web/src/pages/import_export/import/ui.rs +++ b/web/src/pages/import_export/import/ui.rs @@ -90,9 +90,10 @@ pub(crate) fn render_progress(locale: Locale, id: &str, snapshot: &Snapshot) -> }, Stat { label: tr!(locale, "import-progress-rate"), - value: snapshot.rows_per_second().map_or_else(unknown, |rows| { - tr!(locale, "import-progress-rate-value", "rows" => rows as i64) - }), + value: snapshot.rows_per_second().map_or_else( + unknown, + |rows| tr!(locale, "import-progress-rate-value", "rows" => rows as i64), + ), }, Stat { label: tr!(locale, "import-progress-remaining"), @@ -134,24 +135,17 @@ pub(crate) fn render_error(locale: Locale, message: &str) -> String { pub(crate) fn render_import_failure( locale: Locale, - inserted: usize, - prepared_rows: usize, + _inserted: usize, + _prepared_rows: usize, csv_row: usize, backend_message: &str, ) -> String { - let message_key = if inserted == 0 { - "import-failure-no-rows-message" - } else { - "import-failure-message" - }; render(&Alert::error( locale, &tr!(locale, "import-error-title"), &tr!( locale, - message_key, - "inserted" => inserted as i64, - "source_rows" => prepared_rows as i64, + "import-failure-no-rows-message", "row" => csv_row as i64, "error" => backend_message.to_string(), ), @@ -215,10 +209,7 @@ mod tests { destination: strings(&["id:42", "id:57"]), source_position: strings(&["1", "2"]), }, - date_formats: DateFormat::options( - Locale::English, - DateFormat::YearMonthDayFourDigit, - ), + date_formats: DateFormat::options(Locale::English, DateFormat::YearMonthDayFourDigit), step, } } @@ -303,9 +294,18 @@ mod tests { assert!(html.contains("Original table columns"), "{html}"); assert!(html.contains(r#"data-source-position="1""#), "{html}"); assert!(html.contains("value-a"), "{html}"); - assert!(html.contains(r#"name="destination" value="id:42""#), "{html}"); - assert!(html.contains(r#""), "{html}"); + assert!( + html.contains(r#"name="destination" value="id:42""#), + "{html}" + ); + assert!( + html.contains(r#""), + "{html}" + ); } /// The two sides are two lists, not one zipped table: a source chip carries @@ -339,7 +339,10 @@ mod tests { fn a_destination_is_carried_as_its_identity() { let html = render_step(&page(mapping())); - assert!(html.contains(r#"name="destination" value="id:42""#), "{html}"); + assert!( + html.contains(r#"name="destination" value="id:42""#), + "{html}" + ); assert!(!html.contains(r#"name="destination" value="a""#), "{html}"); } @@ -368,11 +371,20 @@ mod tests { html.contains(r#"formaction="/admin/import/prepared.csv""#), "{html}" ); - assert!(html.contains(""a","b","c""), "{html}"); + assert!( + html.contains(""a","b","c""), + "{html}" + ); // The mapping travels with it, so the download and the import prepare // the identical file. - assert!(html.contains(r#"name="destination" value="id:42""#), "{html}"); - assert!(html.contains(r#"name="source_position" value="1""#), "{html}"); + assert!( + html.contains(r#"name="destination" value="id:42""#), + "{html}" + ); + assert!( + html.contains(r#"name="source_position" value="1""#), + "{html}" + ); } /// While it runs, the card says how far along it is and asks for itself @@ -391,10 +403,13 @@ mod tests { }, ); - assert!(html.contains(r#"hx-get="/admin/import/progress/7""#), "{html}"); + assert!( + html.contains(r#"hx-get="/admin/import/progress/7""#), + "{html}" + ); assert!(html.contains(r#"hx-trigger="every 1s""#), "{html}"); assert!(html.contains(r#"value="25""#), "{html}"); - assert!(html.contains("500 of 2000 rows imported"), "{html}"); + assert!(html.contains("500 of 2000 rows staged"), "{html}"); assert!(html.contains("40 s"), "{html}"); // 500 rows in 40 seconds, so the 1500 left are about two minutes away. assert!(html.contains("13 rows/s"), "{html}"); @@ -423,7 +438,7 @@ mod tests { } #[test] - fn an_import_failure_explains_partial_progress_and_the_backend_error() { + fn an_import_failure_explains_atomic_rollback_and_the_backend_error() { let html = render_import_failure( Locale::English, 599, @@ -432,10 +447,10 @@ mod tests { "Internal server error (reference: example-id)", ); - assert!(html.contains("Imported 599 of 1200 rows"), "{html}"); + assert!(html.contains("Nothing was imported"), "{html}"); assert!(html.contains("CSV row 601 failed"), "{html}"); assert!(html.contains("reference: example-id"), "{html}"); - assert!(html.contains("Rows already imported remain"), "{html}"); + assert!(!html.contains("already imported remain"), "{html}"); } #[test]