Merge pull request #65 from syumai/sockets

support TCP sockets

.gitignore

@@ -2,3 +2,4 @@ dist
 build
 node_modules
 .wrangler
+.dev.vars

README.md

@@ -30,6 +30,7 @@
 * [x] Environment variables
 * [x] FetchEvent
 * [x] Cron Triggers
+* [x] TCP Sockets
 
 ## Installation
 

_examples/mysql-blog-server/.dev.vars.example

@@ -0,0 +1 @@
+MYSQL_DSN=user:pass@tcp(hostname)/database-name?interpolateParams=true

_examples/mysql-blog-server/.gitignore

@@ -0,0 +1,3 @@
+build
+node_modules
+.wrangler

_examples/mysql-blog-server/Makefile

@@ -0,0 +1,12 @@
+.PHONY: dev
+dev:
+	wrangler dev
+
+.PHONY: build
+build:
+	go run ../../cmd/workers-assets-gen -mode=go
+	GOOS=js GOARCH=wasm go build -o ./build/app.wasm .
+
+.PHONY: deploy
+deploy:
+	wrangler deploy

_examples/mysql-blog-server/README.md

@@ -0,0 +1,74 @@
+# mysql-blog-server
+
+* A simple Blog server implemented in Go.
+* This example is using MySQL.
+
+# WIP
+
+### Create blog post
+
+```
+$ curl -X POST 'http://localhost:8787/articles' \
+-H 'Content-Type: application/json' \
+-d '{
+  "title":"example post",
+  "body":"body of the example post"
+}'
+{
+  "article": {
+    {
+      "id": "f9e8119e-881e-4dc5-9307-af4f2dc79891",
+      "title": "example post",
+      "body": "body of the example post",
+      "createdAt": 1677382874
+    }
+  }
+}
+```
+
+### List blog posts
+
+```
+$ curl 'http://localhost:8787/articles'
+{
+  "articles": [
+    {
+      "id": "bea6cd80-5a83-45f0-b061-0e13a2ad5fba",
+      "title": "example post 2",
+      "body": "body of the example post 2",
+      "createdAt": 1677383758
+    },
+    {
+      "id": "f9e8119e-881e-4dc5-9307-af4f2dc79891",
+      "title": "example post",
+      "body": "body of the example post",
+      "createdAt": 1677382874
+    }
+  ]
+}
+```
+
+## Development
+
+### Requirements
+
+This project requires these tools to be installed globally.
+
+* wrangler
+* go
+
+### Setup MySQL DB
+
+* This project requires MySQL DB.
+  - Connection setting: `.dev.vars.example` (please rename to `.dev.vars`.)
+  - Initial migration SQL: `schema.sql`
+* If you want to deploy this app to production, please set `MYSQL_DSN` to your Worker secrets.
+  - Run: `npx wrangler secret put MYSQL_DSN`.
+
+### Commands
+
+```
+make dev    # run dev server
+make build  # build Go Wasm binary
+make deploy # deploy worker
+```

_examples/mysql-blog-server/app/handler.go

@@ -0,0 +1,135 @@
+package app
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"fmt"
+	"log"
+	"net"
+	"net/http"
+	"os"
+	"time"
+
+	"github.com/go-sql-driver/mysql"
+	"github.com/syumai/workers/_examples/mysql-blog-server/app/model"
+	"github.com/syumai/workers/cloudflare"
+	"github.com/syumai/workers/cloudflare/sockets"
+)
+
+type articleHandler struct{}
+
+var _ http.Handler = (*articleHandler)(nil)
+
+func NewArticleHandler() http.Handler {
+	return &articleHandler{}
+}
+
+func (h *articleHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
+	// initialize DB.
+	mysql.RegisterDialContext("tcp", func(_ context.Context, addr string) (net.Conn, error) {
+		return sockets.Connect(req.Context(), addr, &sockets.SocketOptions{
+			SecureTransport: sockets.SecureTransportOff,
+		})
+	})
+	db, err := sql.Open("mysql",
+		cloudflare.Getenv(req.Context(), "MYSQL_DSN"))
+	if err != nil {
+		log.Fatalf("failed to connect: %v", err)
+	}
+
+	switch req.Method {
+	case http.MethodGet:
+		h.listArticles(w, req, db)
+		return
+	case http.MethodPost:
+		h.createArticle(w, req, db)
+		return
+	}
+	w.WriteHeader(http.StatusNotFound)
+	w.Header().Set("Content-Type", "text/plain")
+	w.Write([]byte("not found"))
+}
+
+func (h *articleHandler) handleErr(w http.ResponseWriter, status int, msg string) {
+	w.WriteHeader(status)
+	w.Header().Set("Content-Type", "text/plain")
+	w.Write([]byte(msg))
+}
+
+func (h *articleHandler) createArticle(w http.ResponseWriter, req *http.Request, db *sql.DB) {
+	var createArticleReq model.CreateArticleRequest
+	if err := json.NewDecoder(req.Body).Decode(&createArticleReq); err != nil {
+		h.handleErr(w, http.StatusBadRequest,
+			"request format is invalid")
+		return
+	}
+
+	now := time.Now().Unix()
+	article := model.Article{
+		Title:     createArticleReq.Title,
+		Body:      createArticleReq.Body,
+		CreatedAt: uint64(now),
+	}
+
+	result, err := db.Exec(`
+INSERT INTO articles (title, body, created_at)
+VALUES (?, ?, ?)
+   `, article.Title, article.Body, article.CreatedAt)
+	if err != nil {
+		log.Println(err)
+		h.handleErr(w, http.StatusInternalServerError,
+			"failed to save article")
+		return
+	}
+	id, err := result.LastInsertId()
+	if err != nil {
+		log.Println(err)
+		h.handleErr(w, http.StatusInternalServerError,
+			"failed to get ID of inserted article")
+		return
+	}
+	article.ID = uint64(id)
+
+	res := model.CreateArticleResponse{
+		Article: article,
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	if err := json.NewEncoder(w).Encode(res); err != nil {
+		fmt.Fprintf(os.Stderr, "failed to encode response: %w\n", err)
+	}
+}
+
+func (h *articleHandler) listArticles(w http.ResponseWriter, req *http.Request, db *sql.DB) {
+	rows, err := db.Query(`
+SELECT id, title, body, created_at FROM articles
+ORDER BY created_at DESC;
+   `)
+	if err != nil {
+		h.handleErr(w, http.StatusInternalServerError,
+			"failed to load article")
+		return
+	}
+
+	articles := []model.Article{}
+	for rows.Next() {
+		var a model.Article
+		err = rows.Scan(&a.ID, &a.Title, &a.Body, &a.CreatedAt)
+		if err != nil {
+			log.Println(err)
+			h.handleErr(w, http.StatusInternalServerError,
+				"failed to scan article")
+			return
+		}
+		articles = append(articles, a)
+	}
+	res := model.ListArticlesResponse{
+		Articles: articles,
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	if err := json.NewEncoder(w).Encode(res); err != nil {
+		fmt.Fprintf(os.Stderr, "failed to encode response: %w\n", err)
+	}
+}

_examples/mysql-blog-server/app/model/article.go

@@ -0,0 +1,21 @@
+package model
+
+type Article struct {
+	ID        uint64 `json:"id"`
+	Title     string `json:"title"`
+	Body      string `json:"body"`
+	CreatedAt uint64 `json:"createdAt"`
+}
+
+type CreateArticleRequest struct {
+	Title string `json:"title"`
+	Body  string `json:"body"`
+}
+
+type CreateArticleResponse struct {
+	Article Article `json:"article"`
+}
+
+type ListArticlesResponse struct {
+	Articles []Article `json:"articles"`
+}

_examples/mysql-blog-server/go.mod

@@ -0,0 +1,10 @@
+module github.com/syumai/workers/_examples/mysql-blog-server
+
+go 1.21.3
+
+require (
+	github.com/go-sql-driver/mysql v1.7.1
+	github.com/syumai/workers v0.9.0
+)
+
+replace github.com/syumai/workers => ../../

_examples/mysql-blog-server/go.sum

@@ -0,0 +1,2 @@
+github.com/go-sql-driver/mysql v1.7.1 h1:lUIinVbN1DY0xBg0eMOzmmtGoHwWBbvnWubQUrtU8EI=
+github.com/go-sql-driver/mysql v1.7.1/go.mod h1:OXbVy3sEdcQ2Doequ6Z5BW6fXNQTmx+9S1MCJN5yJMI=

_examples/mysql-blog-server/main.go

@@ -0,0 +1,13 @@
+package main
+
+import (
+	"net/http"
+
+	"github.com/syumai/workers"
+	"github.com/syumai/workers/_examples/mysql-blog-server/app"
+)
+
+func main() {
+	http.Handle("/articles", app.NewArticleHandler())
+	workers.Serve(nil) // use http.DefaultServeMux
+}

_examples/mysql-blog-server/schema.sql

@@ -0,0 +1,13 @@
+DROP TABLE IF EXISTS articles;
+CREATE TABLE IF NOT EXISTS articles (
+    id INT PRIMARY KEY AUTO_INCREMENT,
+    title TEXT NOT NULL,
+    body TEXT NOT NULL,
+    created_at INT NOT NULL
+);
+CREATE INDEX idx_articles_on_created_at ON articles (created_at DESC);
+INSERT INTO articles (title, body, created_at) VALUES (
+    'title of example post',
+    'Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.',
+    UNIX_TIMESTAMP()
+);

_examples/mysql-blog-server/wrangler.toml

@@ -0,0 +1,6 @@
+name = "mysql-blog-server"
+main = "./build/worker.mjs"
+compatibility_date = "2024-01-03"
+
+[build]
+command = "make build"

_examples/sockets/.gitignore

@@ -0,0 +1,3 @@
+build
+node_modules
+.wrangler

_examples/sockets/Makefile

@@ -0,0 +1,12 @@
+.PHONY: dev
+dev:
+	wrangler dev
+
+.PHONY: build
+build:
+	go run ../../cmd/workers-assets-gen -mode=go
+	GOOS=js GOARCH=wasm go build -o ./build/app.wasm .
+
+.PHONY: deploy
+deploy:
+	wrangler deploy

_examples/sockets/README.md

@@ -0,0 +1,20 @@
+# sockets
+
+- makes a TCP connection to tcpbin.com:4242 and sends message.
+
+## Development
+
+### Requirements
+
+This project requires these tools to be installed globally.
+
+* wrangler
+* go
+
+### Commands
+
+```
+make dev    # run dev server
+make build  # build Go Wasm binary
+make deploy # deploy worker
+```

_examples/sockets/go.mod

@@ -0,0 +1,7 @@
+module github.com/syumai/workers/_examples/sockets
+
+go 1.21.3
+
+require github.com/syumai/workers v0.0.0
+
+replace github.com/syumai/workers => ../../

_examples/sockets/main.go

@@ -0,0 +1,35 @@
+package main
+
+import (
+	"bufio"
+	"net/http"
+	"time"
+
+	"github.com/syumai/workers"
+	"github.com/syumai/workers/cloudflare/sockets"
+)
+
+func main() {
+	handler := http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		conn, err := sockets.Connect(req.Context(), "tcpbin.com:4242", nil)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+		defer conn.Close()
+		conn.SetDeadline(time.Now().Add(1 * time.Hour))
+		_, err = conn.Write([]byte("hello.\n"))
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+		rd := bufio.NewReader(conn)
+		bts, err := rd.ReadBytes('.')
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+		w.Write(bts)
+	})
+	workers.Serve(handler)
+}

_examples/sockets/wrangler.toml

@@ -0,0 +1,6 @@
+name = "sockets"
+main = "./build/worker.mjs"
+compatibility_date = "2024-01-03"
+
+[build]
+command = "make build"

_templates/cloudflare/pages-tinygo/Makefile

@@ -4,7 +4,7 @@ dev:
 
 .PHONY: build
 build:
-	go run github.com/syumai/workers/cmd/workers-assets-gen@v0.20.0
+	go run github.com/syumai/workers/cmd/workers-assets-gen@v0.21.0
 	tinygo build -o ./build/app.wasm -target wasm -no-debug ./...
 
 .PHONY: deploy

_templates/cloudflare/worker-go/Makefile

@@ -4,7 +4,7 @@ dev:
 
 .PHONY: build
 build:
-	go run github.com/syumai/workers/cmd/workers-assets-gen@v0.20.0 -mode=go
+	go run github.com/syumai/workers/cmd/workers-assets-gen@v0.21.0 -mode=go
 	GOOS=js GOARCH=wasm go build -o ./build/app.wasm .
 
 .PHONY: deploy

_templates/cloudflare/worker-tinygo/Makefile

@@ -4,7 +4,7 @@ dev:
 
 .PHONY: build
 build:
-	go run github.com/syumai/workers/cmd/workers-assets-gen@v0.20.0
+	go run github.com/syumai/workers/cmd/workers-assets-gen@v0.21.0
 	tinygo build -o ./build/app.wasm -target wasm -no-debug ./...
 
 .PHONY: deploy

cloudflare/d1/connector.go

@@ -19,7 +19,7 @@ var (
 // OpenConnector returns Connector of D1.
 // This method checks DB existence. If DB was not found, this function returns error.
 func OpenConnector(ctx context.Context, name string) (driver.Connector, error) {
-	v := cfruntimecontext.GetRuntimeContextEnv(ctx).Get(name)
+	v := cfruntimecontext.MustGetRuntimeContextEnv(ctx).Get(name)
 	if v.IsUndefined() {
 		return nil, ErrDatabaseNotFound
 	}

cloudflare/dostub.go

@@ -21,7 +21,7 @@ type DurableObjectNamespace struct {
 // This binding must be defined in the `wrangler.toml` file. The method will
 // return an `error` when there is no binding defined by `varName`.
 func NewDurableObjectNamespace(ctx context.Context, varName string) (*DurableObjectNamespace, error) {
-	inst := cfruntimecontext.GetRuntimeContextEnv(ctx).Get(varName)
+	inst := cfruntimecontext.MustGetRuntimeContextEnv(ctx).Get(varName)
 	if inst.IsUndefined() {
 		return nil, fmt.Errorf("%s is undefined", varName)
 	}

cloudflare/env.go

@@ -11,12 +11,12 @@ import (
 //   - https://developers.cloudflare.com/workers/platform/environment-variables/
 //   - This function panics when a runtime context is not found.
 func Getenv(ctx context.Context, name string) string {
-	return cfruntimecontext.GetRuntimeContextEnv(ctx).Get(name).String()
+	return cfruntimecontext.MustGetRuntimeContextEnv(ctx).Get(name).String()
 }
 
 // GetBinding gets a value of an environment binding.
 //   - https://developers.cloudflare.com/workers/platform/bindings/about-service-bindings/
 //   - This function panics when a runtime context is not found.
 func GetBinding(ctx context.Context, name string) js.Value {
-	return cfruntimecontext.GetRuntimeContextEnv(ctx).Get(name)
+	return cfruntimecontext.MustGetRuntimeContextEnv(ctx).Get(name)
 }

cloudflare/fetchevent.go

@@ -12,7 +12,7 @@ import (
 // It accepts an asynchronous task which the Workers runtime will execute before the handler terminates but without blocking the response.
 // see: https://developers.cloudflare.com/workers/runtime-apis/fetch-event/#waituntil
 func WaitUntil(ctx context.Context, task func()) {
-	exCtx := cfruntimecontext.GetExecutionContext(ctx)
+	exCtx := cfruntimecontext.MustGetExecutionContext(ctx)
 	exCtx.Call("waitUntil", jsutil.NewPromise(js.FuncOf(func(this js.Value, pArgs []js.Value) any {
 		resolve := pArgs[0]
 		go func() {
@@ -27,7 +27,7 @@ func WaitUntil(ctx context.Context, task func()) {
 // Instead, the request forwards to the origin server as if it had not gone through the worker.
 // see: https://developers.cloudflare.com/workers/runtime-apis/fetch-event/#passthroughonexception
 func PassThroughOnException(ctx context.Context) {
-	exCtx := cfruntimecontext.GetExecutionContext(ctx)
+	exCtx := cfruntimecontext.MustGetExecutionContext(ctx)
 	jsutil.AwaitPromise(jsutil.NewPromise(js.FuncOf(func(this js.Value, pArgs []js.Value) any {
 		resolve := pArgs[0]
 		go func() {

cloudflare/internal/cfruntimecontext/cfruntimecontext.go

@@ -2,6 +2,7 @@ package cfruntimecontext
 
 import (
 	"context"
+	"errors"
 	"syscall/js"
 
 	"github.com/syumai/workers/internal/runtimecontext"
@@ -13,23 +14,45 @@ import (
  *  type RuntimeContext {
  *    env: Env;
  *    ctx: ExecutionContext;
+ *    ...
  *  }
  *  ```
  * This type is based on the type definition of ExportedHandlerFetchHandler.
  * - see: https://github.com/cloudflare/workers-types/blob/c8d9533caa4415c2156d2cf1daca75289d01ae70/index.d.ts#LL564
  */
 
-// GetRuntimeContextEnv gets object which holds environment variables bound to Cloudflare worker.
+// MustGetRuntimeContextEnv gets object which holds environment variables bound to Cloudflare worker.
 // - see: https://github.com/cloudflare/workers-types/blob/c8d9533caa4415c2156d2cf1daca75289d01ae70/index.d.ts#L566
-func GetRuntimeContextEnv(ctx context.Context) js.Value {
-	runtimeCtxValue := runtimecontext.MustExtract(ctx)
-	return runtimeCtxValue.Get("env")
+func MustGetRuntimeContextEnv(ctx context.Context) js.Value {
+	return MustGetRuntimeContextValue(ctx, "env")
 }
 
-// GetExecutionContext gets ExecutionContext object from context.
+// MustGetExecutionContext gets ExecutionContext object from context.
 // - see: https://github.com/cloudflare/workers-types/blob/c8d9533caa4415c2156d2cf1daca75289d01ae70/index.d.ts#L567
 // - see also: https://github.com/cloudflare/workers-types/blob/c8d9533caa4415c2156d2cf1daca75289d01ae70/index.d.ts#L554
-func GetExecutionContext(ctx context.Context) js.Value {
-	runtimeCtxValue := runtimecontext.MustExtract(ctx)
-	return runtimeCtxValue.Get("ctx")
+func MustGetExecutionContext(ctx context.Context) js.Value {
+	return MustGetRuntimeContextValue(ctx, "ctx")
+}
+
+// MustGetRuntimeContextValue gets value for specified key from RuntimeContext.
+// - if the value is undefined, this function panics.
+func MustGetRuntimeContextValue(ctx context.Context, key string) js.Value {
+	val, err := GetRuntimeContextValue(ctx, key)
+	if err != nil {
+		panic(err)
+	}
+	return val
+}
+
+var ErrValueNotFound = errors.New("execution context value for specified key not found")
+
+// GetRuntimeContextValue gets value for specified key from RuntimeContext.
+// - if the value is undefined, return error.
+func GetRuntimeContextValue(ctx context.Context, key string) (js.Value, error) {
+	runtimeCtxValue := runtimecontext.MustExtract(ctx)
+	v := runtimeCtxValue.Get(key)
+	if v.IsUndefined() {
+		return js.Value{}, ErrValueNotFound
+	}
+	return v, nil
 }

cloudflare/kv.go

@@ -23,7 +23,7 @@ type KVNamespace struct {
 //   - if the given variable name doesn't exist on runtime context, returns error.
 //   - This function panics when a runtime context is not found.
 func NewKVNamespace(ctx context.Context, varName string) (*KVNamespace, error) {
-	inst := cfruntimecontext.GetRuntimeContextEnv(ctx).Get(varName)
+	inst := cfruntimecontext.MustGetRuntimeContextEnv(ctx).Get(varName)
 	if inst.IsUndefined() {
 		return nil, fmt.Errorf("%s is undefined", varName)
 	}

cloudflare/r2bucket.go

@@ -23,7 +23,7 @@ type R2Bucket struct {
 //   - if the given variable name doesn't exist on runtime context, returns error.
 //   - This function panics when a runtime context is not found.
 func NewR2Bucket(ctx context.Context, varName string) (*R2Bucket, error) {
-	inst := cfruntimecontext.GetRuntimeContextEnv(ctx).Get(varName)
+	inst := cfruntimecontext.MustGetRuntimeContextEnv(ctx).Get(varName)
 	if inst.IsUndefined() {
 		return nil, fmt.Errorf("%s is undefined", varName)
 	}

cloudflare/sockets/connect.go

@@ -0,0 +1,54 @@
+package sockets
+
+import (
+	"context"
+	"net"
+	"syscall/js"
+	"time"
+
+	"github.com/syumai/workers/cloudflare/internal/cfruntimecontext"
+	"github.com/syumai/workers/internal/jsutil"
+)
+
+type SecureTransport string
+
+const (
+	// SecureTransportOn indicates "Use TLS".
+	SecureTransportOn SecureTransport = "on"
+	// SecureTransportOff indicates "Do not use TLS".
+	SecureTransportOff SecureTransport = "off"
+	// SecureTransportStartTLS indicates "Do not use TLS initially, but allow the socket to be upgraded
+	// to use TLS by calling *Socket.StartTLS()".
+	SecureTransportStartTLS SecureTransport = "starttls"
+)
+
+type SocketOptions struct {
+	SecureTransport SecureTransport `json:"secureTransport"`
+	AllowHalfOpen   bool            `json:"allowHalfOpen"`
+}
+
+const defaultDeadline = 999999 * time.Hour
+
+func Connect(ctx context.Context, addr string, opts *SocketOptions) (net.Conn, error) {
+	connect, err := cfruntimecontext.GetRuntimeContextValue(ctx, "connect")
+	if err != nil {
+		return nil, err
+	}
+	optionsObj := jsutil.NewObject()
+	if opts != nil {
+		if opts.AllowHalfOpen {
+			optionsObj.Set("allowHalfOpen", true)
+		}
+		if opts.SecureTransport != "" {
+			optionsObj.Set("secureTransport", string(opts.SecureTransport))
+		}
+	}
+	sockVal, err := jsutil.TryCatch(js.FuncOf(func(_ js.Value, args []js.Value) any {
+		return connect.Invoke(addr, optionsObj)
+	}))
+	if err != nil {
+		return nil, err
+	}
+	deadline := time.Now().Add(defaultDeadline)
+	return newSocket(ctx, sockVal, deadline, deadline), nil
+}

cloudflare/sockets/socket.go

@@ -0,0 +1,176 @@
+package sockets
+
+import (
+	"context"
+	"io"
+	"net"
+	"os"
+	"syscall/js"
+	"time"
+
+	"github.com/syumai/workers/internal/jsutil"
+)
+
+func newSocket(ctx context.Context, sockVal js.Value, readDeadline, writeDeadline time.Time) *Socket {
+	ctx, cancel := context.WithCancel(ctx)
+	writerVal := sockVal.Get("writable").Call("getWriter")
+	readerVal := sockVal.Get("readable").Call("getReader")
+	return &Socket{
+		ctx:    ctx,
+		cancel: cancel,
+
+		reader:    jsutil.ConvertStreamReaderToReader(readerVal),
+		writerVal: writerVal,
+
+		readDeadline:  readDeadline,
+		writeDeadline: writeDeadline,
+
+		startTLS:   func() js.Value { return sockVal.Call("startTls") },
+		close:      func() { sockVal.Call("close") },
+		closeRead:  func() { readerVal.Call("close") },
+		closeWrite: func() { writerVal.Call("close") },
+	}
+}
+
+type Socket struct {
+	ctx    context.Context
+	cancel context.CancelFunc
+
+	reader    io.Reader
+	writerVal js.Value
+
+	readDeadline  time.Time
+	writeDeadline time.Time
+
+	startTLS   func() js.Value
+	close      func()
+	closeRead  func()
+	closeWrite func()
+}
+
+var _ net.Conn = (*Socket)(nil)
+
+// Read reads data from the connection.
+// Read can be made to time out and return an error after a fixed
+// time limit; see SetDeadline and SetReadDeadline.
+func (t *Socket) Read(b []byte) (n int, err error) {
+	ctx, cancel := context.WithDeadline(t.ctx, t.readDeadline)
+	defer cancel()
+	done := make(chan struct{})
+	go func() {
+		n, err = t.reader.Read(b)
+		close(done)
+	}()
+	select {
+	case <-done:
+		return
+	case <-ctx.Done():
+		return 0, os.ErrDeadlineExceeded
+	}
+}
+
+// Write writes data to the connection.
+// Write can be made to time out and return an error after a fixed
+// time limit; see SetDeadline and SetWriteDeadline.
+func (t *Socket) Write(b []byte) (n int, err error) {
+	ctx, cancel := context.WithDeadline(t.ctx, t.writeDeadline)
+	defer cancel()
+	done := make(chan struct{})
+	go func() {
+		arr := jsutil.NewUint8Array(len(b))
+		js.CopyBytesToJS(arr, b)
+		_, err = jsutil.AwaitPromise(t.writerVal.Call("write", arr))
+		// TODO: handle error
+		if err == nil {
+			n = len(b)
+		}
+		close(done)
+	}()
+	select {
+	case <-done:
+		return
+	case <-ctx.Done():
+		return 0, os.ErrDeadlineExceeded
+	}
+}
+
+// StartTLS upgrades an insecure socket to a secure one that uses TLS, returning a new *Socket.
+
+func (t *Socket) StartTLS() *Socket {
+	return newSocket(t.ctx, t.startTLS(), t.readDeadline, t.writeDeadline)
+}
+
+// Close closes the connection.
+// Any blocked Read or Write operations will be unblocked and return errors.
+func (t *Socket) Close() error {
+	defer t.cancel()
+	t.close()
+	return nil
+}
+
+// CloseRead closes the read side of the connection.
+func (t *Socket) CloseRead() error {
+	t.closeRead()
+	return nil
+}
+
+// CloseWrite closes the write side of the connection.
+func (t *Socket) CloseWrite() error {
+	t.closeWrite()
+	return nil
+}
+
+// LocalAddr returns the local network address, if known.
+func (t *Socket) LocalAddr() net.Addr {
+	return nil
+}
+
+// RemoteAddr returns the remote network address, if known.
+func (t *Socket) RemoteAddr() net.Addr {
+	return nil
+}
+
+// SetDeadline sets the read and write deadlines associated
+// with the connection. It is equivalent to calling both
+// SetReadDeadline and SetWriteDeadline.
+//
+// A deadline is an absolute time after which I/O operations
+// fail instead of blocking. The deadline applies to all future
+// and pending I/O, not just the immediately following call to
+// Read or Write. After a deadline has been exceeded, the
+// connection can be refreshed by setting a deadline in the future.
+//
+// If the deadline is exceeded a call to Read or Write or to other
+// I/O methods will return an error that wraps os.ErrDeadlineExceeded.
+// This can be tested using errors.Is(err, os.ErrDeadlineExceeded).
+// The error's Timeout method will return true, but note that there
+// are other possible errors for which the Timeout method will
+// return true even if the deadline has not been exceeded.
+//
+// An idle timeout can be implemented by repeatedly extending
+// the deadline after successful Read or Write calls.
+//
+// A zero value for t means I/O operations will not time out.
+func (t *Socket) SetDeadline(deadline time.Time) error {
+	t.SetReadDeadline(deadline)
+	t.SetWriteDeadline(deadline)
+	return nil
+}
+
+// SetReadDeadline sets the deadline for future Read calls
+// and any currently-blocked Read call.
+// A zero value for t means Read will not time out.
+func (t *Socket) SetReadDeadline(deadline time.Time) error {
+	t.readDeadline = deadline
+	return nil
+}
+
+// SetWriteDeadline sets the deadline for future Write calls
+// and any currently-blocked Write call.
+// Even if write times out, it may return n > 0, indicating that
+// some of the data was successfully written.
+// A zero value for t means Write will not time out.
+func (t *Socket) SetWriteDeadline(deadline time.Time) error {
+	t.writeDeadline = deadline
+	return nil
+}

cmd/workers-assets-gen/assets/common/shim.mjs

@@ -1,9 +1,22 @@
 import "./wasm_exec.js";
+import { connect } from 'cloudflare:sockets';
 
 const go = new Go();
 
 let mod;
 
+globalThis.tryCatch = (fn) => {
+  try {
+    return {
+      result: fn(),
+    };
+  } catch(e) {
+    return {
+      error: e,
+    }
+  }
+}
+
 export function init(m) {
   mod = m;
 }
@@ -17,19 +30,27 @@ async function run() {
   await readyPromise;
 }
 
+function createRuntimeContext(env, ctx) {
+  return {
+    env,
+    ctx,
+    connect,
+  }
+}
+
 export async function fetch(req, env, ctx) {
   await run();
-  return handleRequest(req, { env, ctx });
+  return handleRequest(req, createRuntimeContext(env, ctx));
 }
 
 export async function scheduled(event, env, ctx) {
   await run();
-  return runScheduler(event, { env, ctx });
+  return runScheduler(event, createRuntimeContext(env, ctx));
 }
 
 // onRequest handles request to Cloudflare Pages
 export async function onRequest(ctx) {
   await run();
   const { request, env } = ctx;
-  return handleRequest(request, { env, ctx });
+  return handleRequest(request, createRuntimeContext(env, ctx));
 }

internal/jsutil/trycatch.go

@@ -0,0 +1,16 @@
+package jsutil
+
+import (
+	"errors"
+	"syscall/js"
+)
+
+func TryCatch(fn js.Func) (js.Value, error) {
+	fnResultVal := js.Global().Call("tryCatch", fn)
+	resultVal := fnResultVal.Get("result")
+	errorVal := fnResultVal.Get("error")
+	if !errorVal.IsUndefined() {
+		return js.Value{}, errors.New(errorVal.String())
+	}
+	return resultVal, nil
+}