fix(bindgen): flat stream lift for nested streams

Author: vados-cosmonic

crates/js-component-bindgen/src/intrinsics/lift.rs

@@ -1017,43 +1017,65 @@ impl LiftIntrinsic {
                     Intrinsic::AsyncStream(AsyncStreamIntrinsic::ExternalStreamClass).name();
                 let lift_flat_stream_fn = self.name();
                 let global_stream_table_map = AsyncStreamIntrinsic::GlobalStreamTableMap.name();
+                let lift_u32 = Self::LiftFlatU32.name();
 
                 output.push_str(&format!(r#"
-                    function {lift_flat_stream_fn}(streamTableIdx, ctx) {{
-                        {debug_log_fn}('[{lift_flat_stream_fn}()] args', {{ streamTableIdx, ctx }});
-                        const {{ memory, useDirectParams, params }} = ctx;
+                    function {lift_flat_stream_fn}(meta) {{
+                        const {{
+                            streamTableIdx,
+                            componentIdx,
+                            isBorrowedType,
+                            isNoneType,
+                            isNumericTypeJs,
+                        }} = meta;
+
+                        return function {lift_flat_stream_fn}Inner(ctx) {{
+                            {debug_log_fn}('[{lift_flat_stream_fn}()] args', {{ ctx }});
+
+                            const streamMeta = {global_stream_table_map}[streamTableIdx];
+                            if (streamMeta.componentIdx !== componentIdx) {{
+                                throw new Error('unexpectedly mismatched component idx');
+                            }}
+                            const {{ table }} = streamMeta;
+                            if (componentIdx === undefined || !table) {{
+                                throw new Error(`invalid global stream table state for table [${{tableIdx}}]`);
+                            }}
 
-                        const {{ table, componentIdx }} = {global_stream_table_map}[streamTableIdx];
-                        if (componentIdx === undefined || !table) {{
-                            throw new Error(`invalid global stream table state for table [${{tableIdx}}]`);
-                        }}
+                            let streamEndWaitableIdx;
+                            if (ctx.useDirectParams) {{
+                                streamEndWaitableIdx = ctx.params[0];
+                                ctx.params = ctx.params.slice(1);
+                            }} else {{
+                                const [waitableIdx, newCtx] = {lift_u32}(ctx);
+                                ctx = newCtx;
+                                streamEndWaitableIdx = waitableIdx;
+                            }}
 
-                        const cstate = {get_or_create_async_state_fn}(componentIdx);
-                        if (!cstate) {{ throw new Error(`missing async state for component [${{componentIdx}}]`); }}
+                            if (!streamEndWaitableIdx) {{ throw new Error('missing stream idx'); }}
 
-                        const streamEndWaitableIdx = params[0];
-                        if (!streamEndWaitableIdx) {{ throw new Error('missing stream idx'); }}
+                            const cstate = {get_or_create_async_state_fn}(componentIdx);
+                            if (!cstate) {{ throw new Error(`missing async state for component [${{componentIdx}}]`); }}
 
-                        const streamEnd = cstate.getStreamEnd({{ tableIdx: streamTableIdx, streamEndWaitableIdx }});
-                        if (!streamEnd) {{
-                            throw new Error(`missing stream end [${{streamEndWaitableIdx}}] (table [${{streamTableIdx}}]) in component [${{componentIdx}}] during lift`);
-                        }}
+                            const streamEnd = cstate.getStreamEnd({{ tableIdx: streamTableIdx, streamEndWaitableIdx }});
+                            if (!streamEnd) {{
+                                throw new Error(`missing stream end [${{streamEndWaitableIdx}}] (table [${{streamTableIdx}}]) in component [${{componentIdx}}] during lift`);
+                            }}
 
-                        // TODO: check for borrowed type
-                        // TODO: check for readable only
-                        // TODO: confirm shared type matches tyep for lift
-                        // TODO: check for IDLE state
-
-                        const stream = new {external_stream_class}({{
-                            globalRep: streamEnd.globalStreamMapRep(),
-                            isReadable: streamEnd.isReadable(),
-                            isWritable: streamEnd.isWritable(),
-                            writeFn: (v) => {{ return streamEnd.write(v); }},
-                            readFn: () => {{ return streamEnd.read(); }},
-                            dropFn: () => {{ return streamEnd.drop(); }},
-                        }});
-
-                        return [ stream, ctx ];
+                            if (ctx.isBorrowed) {{ throw new Error('cannot lift flat stream of borrowed type'); }}
+                            if (streamEnd.isWritable()) {{ throw new Error('only readable streams can be lifted'); }}
+                            if (!streamEnd.isIdle()) {{ throw new Error('streams must be in idle state'); }}
+
+                            const stream = new {external_stream_class}({{
+                                globalRep: streamEnd.globalStreamMapRep(),
+                                isReadable: streamEnd.isReadable(),
+                                isWritable: streamEnd.isWritable(),
+                                writeFn: (v) => {{ return streamEnd.write(v); }},
+                                readFn: () => {{ return streamEnd.read(); }},
+                                dropFn: () => {{ return streamEnd.drop(); }},
+                            }});
+
+                            return [ stream, ctx ];
+                        }}
                     }}
                 "#));
             }

crates/js-component-bindgen/src/intrinsics/p3/async_stream.rs

@@ -1006,6 +1006,7 @@ impl AsyncStreamIntrinsic {
                         isDone() {{ return this.getCopyState() === {stream_end_class}.CopyState.DONE; }}
                         isCompleted() {{ return this.getCopyState() === {stream_end_class}.CopyState.COMPLETED; }}
                         isDropped() {{ return this.getCopyState() === {stream_end_class}.CopyState.DROPPED; }}
+                        isIdle() {{ return this.getCopyState() === {stream_end_class}.CopyState.IDLE; }}
 
                         {action_impl}
                         {inner_rw_impl}

crates/js-component-bindgen/src/transpile_bindgen.rs

@@ -4897,7 +4897,41 @@ pub fn gen_flat_lift_fn_js_expr(
             intrinsic_mgr.add_intrinsic(Intrinsic::Lift(LiftIntrinsic::LiftFlatStream));
             let table_idx = ty_idx.as_u32();
             let f = Intrinsic::Lift(LiftIntrinsic::LiftFlatStream).name();
-            format!("{f}.bind(null, {table_idx})")
+            let table_ty = &component_types[*ty_idx];
+            let component_idx = table_ty.instance.as_u32();
+            let stream_ty_idx = table_ty.ty;
+            let stream_ty = &component_types[stream_ty_idx];
+            let payload = stream_ty.payload;
+            let (is_borrowed, is_none_type_js, is_numeric_type_js) = match payload {
+                None => (false, true, false),
+                Some(t) => (
+                    matches!(t, InterfaceType::Borrow(_)),
+                    false,
+                    matches!(
+                        ty,
+                        InterfaceType::U8
+                            | InterfaceType::U16
+                            | InterfaceType::U32
+                            | InterfaceType::U64
+                            | InterfaceType::S8
+                            | InterfaceType::S16
+                            | InterfaceType::S32
+                            | InterfaceType::S64
+                            | InterfaceType::Float32
+                            | InterfaceType::Float64
+                    ),
+                ),
+            };
+
+            format!(
+                r#"{f}({{
+                 streamTableIdx: {table_idx},
+                 componentIdx: {component_idx},
+                 isBorrowedType: {is_borrowed},
+                 isNoneType: {is_none_type_js},
+                 isNumericTypeJs: {is_numeric_type_js},
+             }})"#
+            )
         }
 
         InterfaceType::ErrorContext(ty_idx) => {