smallweb
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561package worker
import ( "bufio" "context" _ "embed" "encoding/base64" "encoding/json" "fmt" "io" "log/slog" "net" "net/http" "os" "os/exec" "path/filepath" "strings" "sync" "sync/atomic" "time"
"github.com/adrg/xdg" "github.com/gorilla/websocket" "github.com/pomdtr/smallweb/internal/app" "github.com/pomdtr/smallweb/internal/build" "github.com/pomdtr/smallweb/internal/utils")
var sandboxUrl = "jsr:@smallweb/sandbox@0.1.1"
type Worker struct { App app.App Env map[string]string StartedAt time.Time Logger *slog.Logger Admin bool
port int idleTimer *time.Timer command *exec.Cmd activeRequests atomic.Int32}
func commandEnv(a app.App, admin bool) []string { env := []string{}
for k, v := range a.Env { env = append(env, fmt.Sprintf("%s=%s", k, v)) }
env = append(env, fmt.Sprintf("HOME=%s", os.Getenv("HOME"))) env = append(env, "DENO_NO_UPDATE_CHECK=1")
env = append(env, fmt.Sprintf("SMALLWEB_VERSION=%s", build.Version)) env = append(env, fmt.Sprintf("SMALLWEB_DIR=%s", a.RootDir)) env = append(env, fmt.Sprintf("SMALLWEB_DOMAIN=%s", a.RootDomain)) env = append(env, fmt.Sprintf("SMALLWEB_APP_NAME=%s", a.Name)) env = append(env, fmt.Sprintf("SMALLWEB_APP_DOMAIN=%s", a.Domain)) env = append(env, fmt.Sprintf("SMALLWEB_APP_URL=%s", a.URL)) env = append(env, fmt.Sprintf("SMALLWEB_APP_DIR=%s", a.BaseDir)) if admin { env = append(env, "SMALLWEB_ADMIN=1") }
// open telemetry for _, value := range os.Environ() { if strings.HasPrefix(value, "OTEL_") { env = append(env, value) } } env = append(env, fmt.Sprintf("OTEL_SERVICE_NAME=%s", a.Domain))
return env}
func NewWorker(app app.App, admin bool, logger *slog.Logger) *Worker { worker := &Worker{ App: app, Logger: logger, Admin: admin, }
return worker}
var upgrader = websocket.Upgrader{} // use default options
type SandboxMethod string
func (me *Worker) DenoArgs(deno string) ([]string, error) { args := []string{ "--allow-net", "--allow-import", "--allow-env", "--allow-sys", "--unstable-kv", "--unstable-otel", "--unstable-temporal", "--node-modules-dir=none", "--no-prompt", "--quiet", }
npmCache := filepath.Join(xdg.CacheHome, "deno", "npm", "registry.npmjs.org") if me.Admin { args = append( args, fmt.Sprintf("--allow-read=%s,%s,%s", me.App.RootDir, deno, npmCache), fmt.Sprintf("--allow-write=%s", me.App.RootDir), )
return args, nil }
for _, configName := range []string{"deno.json", "deno.jsonc"} { configPath := filepath.Join(me.App.Dir(), configName) if _, err := os.Stat(configPath); err == nil { args = append(args, fmt.Sprintf("--config=%s", configPath)) break } }
// if root is not a symlink appDir := me.App.Dir() if fi, err := os.Lstat(appDir); err == nil && fi.Mode()&os.ModeSymlink == 0 { args = append( args, fmt.Sprintf("--allow-read=%s,%s,%s", appDir, deno, npmCache), fmt.Sprintf("--allow-write=%s", me.App.DataDir()), )
return args, nil }
target, err := os.Readlink(appDir) if err != nil { return nil, fmt.Errorf("could not read symlink: %w", err) }
if !filepath.IsAbs(target) { target = filepath.Join(filepath.Dir(appDir), target) }
args = append( args, fmt.Sprintf("--allow-read=%s,%s,%s,%s", appDir, target, deno, npmCache), fmt.Sprintf("--allow-write=%s,%s", me.App.DataDir(), filepath.Join(target, "data")), )
return args, nil}
func (me *Worker) Start() error { port, err := GetFreePort() if err != nil { return fmt.Errorf("could not get free port: %w", err) } me.port = port
deno, err := DenoExecutable() if err != nil { return fmt.Errorf("could not find deno executable") }
args := []string{"run"} denoArgs, err := me.DenoArgs(deno) if err != nil { return fmt.Errorf("could not get deno args: %w", err) }
args = append(args, denoArgs...) input := strings.Builder{} encoder := json.NewEncoder(&input) encoder.SetEscapeHTML(false) if err := encoder.Encode(map[string]any{ "command": "fetch", "entrypoint": me.App.Entrypoint(), "port": port, }); err != nil { return fmt.Errorf("could not encode input: %w", err) }
args = append(args, sandboxUrl, input.String())
command := exec.Command(deno, args...) command.Dir = me.App.Dir() command.Env = commandEnv(me.App, me.Admin)
stdoutPipe, err := command.StdoutPipe() if err != nil { return fmt.Errorf("could not get stdout pipe: %w", err) }
stderrPipe, err := command.StderrPipe() if err != nil { return fmt.Errorf("could not get stderr pipe: %w", err) }
if err := command.Start(); err != nil { return fmt.Errorf("could not start server: %w", err) }
readyChan := make(chan bool) go func() { scanner := bufio.NewScanner(stderrPipe) for scanner.Scan() { if scanner.Text() == "READY" { readyChan <- true return }
os.Stderr.WriteString(scanner.Text() + "\n") }
readyChan <- false }()
select { case ready := <-readyChan: if !ready { return fmt.Errorf("server did not start correctly") } case <-time.After(30 * time.Second): return fmt.Errorf("server start timed out") }
// Function to handle logging for both stdout and stderr logPipe := func(pipe io.ReadCloser, stream string) { scanner := bufio.NewScanner(pipe) for scanner.Scan() { if me.Logger == nil { if stream == "stderr" { fmt.Fprintln(os.Stderr, scanner.Text()) } else { fmt.Fprintln(os.Stdout, scanner.Text()) } continue } me.Logger.Info( scanner.Text(), "stream", stream, ) } }
// Start goroutine for stdout go logPipe(stdoutPipe, "stdout")
// Start goroutine for stderr go logPipe(stderrPipe, "stderr")
me.command = command me.StartedAt = time.Now() me.idleTimer = time.NewTimer(10 * time.Second) go me.monitorIdleTimer()
return nil}
func (me *Worker) IsRunning() bool { return me.command != nil}
func (me *Worker) Stop() error { if !me.IsRunning() { return nil }
command := me.command me.command = nil
if err := command.Process.Signal(os.Interrupt); err != nil { return fmt.Errorf("failed to send interrupt signal: %w", err) }
done := make(chan error, 1) go func() { done <- command.Wait() }()
select { case <-time.After(5 * time.Second): if err := command.Process.Kill(); err != nil { return fmt.Errorf("failed to kill process: %w", err) } return fmt.Errorf("process did not exit after 5 seconds") case <-done: return nil }}
func (me *Worker) monitorIdleTimer() { for { <-me.idleTimer.C // if there are no active requests, stop the worker if me.activeRequests.Load() == 0 { _ = me.Stop() return } }}
func (me *Worker) ServeHTTP(w http.ResponseWriter, r *http.Request) { me.activeRequests.Add(1) defer func() { me.idleTimer.Reset(10 * time.Second) me.activeRequests.Add(-1) }()
// handle websockets if r.Header.Get("Upgrade") == "websocket" { serverConn, err := upgrader.Upgrade(w, r, nil) if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return } defer serverConn.Close()
clientConn, _, err := websocket.DefaultDialer.Dial(fmt.Sprintf("ws://127.0.0.1:%d%s", me.port, r.URL.Path), nil) if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return } defer clientConn.Close()
ctx, cancel := context.WithCancel(r.Context()) defer cancel()
// Channel to signal closure var wg sync.WaitGroup wg.Add(2) go func() { defer wg.Done() for { select { case <-ctx.Done(): return default: messageType, p, err := clientConn.ReadMessage() if err != nil { return }
if err := serverConn.WriteMessage(messageType, p); err != nil { return } } } }()
go func() { defer wg.Done() for { select { case <-ctx.Done(): return default: messageType, p, err := serverConn.ReadMessage() if err != nil { return }
if err := clientConn.WriteMessage(messageType, p); err != nil { return } } } }()
wg.Wait() return }
request, err := http.NewRequestWithContext(r.Context(), r.Method, fmt.Sprintf("http://127.0.0.1:%d%s", me.port, r.URL.String()), r.Body) if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return }
for k, v := range r.Header { for _, vv := range v { request.Header.Add(k, vv) } }
if request.Header.Get("x-forwarded-host") == "" { request.Header.Set("x-forwarded-host", r.Host) }
if request.Header.Get("x-forwarded-proto") == "" { request.Header.Set("x-forwarded-proto", "https") }
client := &http.Client{ CheckRedirect: func(req *http.Request, via []*http.Request) error { return http.ErrUseLastResponse }, Timeout: 5 * time.Minute, }
resp, err := client.Do(request) if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return } defer resp.Body.Close()
for k, v := range resp.Header { for _, vv := range v { w.Header().Add(k, vv) } }
w.WriteHeader(resp.StatusCode)
flusher, ok := w.(http.Flusher) if !ok { http.Error(w, "expected http.ResponseWriter to be an http.Flusher", http.StatusInternalServerError) return }
// Stream the response body to the client buf := make([]byte, 1024) for { n, err := resp.Body.Read(buf) if n > 0 { _, writeErr := w.Write(buf[:n]) if writeErr != nil { return } flusher.Flush() // flush the buffer to the client } if err != nil { if err != io.EOF { http.Error(w, "Error reading response body", http.StatusInternalServerError) } break } }}
func DenoExecutable() (string, error) { if env, ok := os.LookupEnv("DENO_EXEC_PATH"); ok { return env, nil }
if denoPath, err := exec.LookPath("deno"); err == nil { return denoPath, nil }
homedir, err := os.UserHomeDir() if err != nil { return "", err }
for _, candidate := range []string{ filepath.Join(homedir, ".deno", "bin", "deno"), "/home/linuxbrew/.linuxbrew/bin/deno", "/opt/homebrew/bin/deno", "/usr/local/bin/deno", "/usr/bin/deno", } { if utils.FileExists(candidate) { return candidate, nil } }
return "", fmt.Errorf("deno executable not found")}
func (me *Worker) Command(ctx context.Context, args []string) (*exec.Cmd, error) { if args == nil { args = []string{} }
deno, err := DenoExecutable() if err != nil { return nil, fmt.Errorf("could not find deno executable") }
cmdArgs := []string{"run"} denoArgs, err := me.DenoArgs(deno) if err != nil { return nil, fmt.Errorf("could not get deno args: %w", err) } cmdArgs = append(cmdArgs, denoArgs...)
payload := strings.Builder{} encoder := json.NewEncoder(&payload) encoder.SetEscapeHTML(false) if err := encoder.Encode(map[string]any{ "command": "run", "entrypoint": me.App.Entrypoint(), "args": args, }); err != nil { return nil, fmt.Errorf("could not encode input: %w", err) }
cmdArgs = append(cmdArgs, sandboxUrl, payload.String())
command := exec.CommandContext(ctx, deno, cmdArgs...) command.Dir = me.App.Dir()
command.Env = commandEnv(me.App, me.Admin)
return command, nil}
func (me *Worker) SendEmail(ctx context.Context, msg []byte) error { deno, err := DenoExecutable() if err != nil { return fmt.Errorf("could not find deno executable") }
args := []string{"run"} denoArgs, err := me.DenoArgs(deno) if err != nil { return fmt.Errorf("could not get deno args: %w", err) }
args = append(args, denoArgs...)
payload := strings.Builder{} encoder := json.NewEncoder(&payload) encoder.SetEscapeHTML(false) if err := encoder.Encode(map[string]any{ "command": "email", "entrypoint": me.App.Entrypoint(), "msg": base64.StdEncoding.EncodeToString(msg), }); err != nil { return fmt.Errorf("could not encode input: %w", err) }
denoArgs = append(args, sandboxUrl, payload.String())
command := exec.CommandContext(ctx, deno, denoArgs...)
command.Stderr = os.Stderr command.Stdout = os.Stdout command.Dir = me.App.Dir()
command.Env = commandEnv(me.App, me.Admin)
return command.Run()}
// GetFreePort asks the kernel for a free open port that is ready to use.func GetFreePort() (int, error) { addr, err := net.ResolveTCPAddr("tcp", "localhost:0") if err != nil { return 0, err }
l, err := net.ListenTCP("tcp", addr) if err != nil { return 0, err } defer l.Close() return l.Addr().(*net.TCPAddr).Port, nil}