Go source walkthrough

The runnable source below is included from the same files packaged in the notes download. Lockfiles, configuration, and the shared browser UI are included in the download.

Database

A notes request reaches this adapter after input validation. DATABASE_ENGINE chooses PostgreSQL, MariaDB, or MongoDB; DATABASE_URL comes from the app’s attachment. SQL uses parameters rather than interpolating note text, and MongoDB uses driver operations. The initialization process creates a table or unique index. Records live in the database, so replacing a web instance leaves them available to its replacement.

func database(ctx context.Context) error {
	dbMu.Lock()
	defer dbMu.Unlock()
	if db != nil || collection != nil {
		return nil
	}
	raw := os.Getenv("DATABASE_URL")
	if raw == "" {
		return errors.New("attach a database first")
	}
	u, err := url.Parse(raw)
	if err != nil {
		return err
	}
	switch engine {
	case "mongodb":
		client, err := mongo.Connect(ctx, options.Client().ApplyURI(raw).SetServerSelectionTimeout(5*time.Second))
		if err != nil {
			return err
		}
		collection = client.Database(strings.TrimPrefix(u.Path, "/")).Collection("notes")
	case "postgresql":
		db, err = sql.Open("pgx", raw)
	case "mariadb":
		c := mysql.NewConfig()
		c.Net = "tcp"
		c.Addr = net.JoinHostPort(u.Hostname(), envPort(u.Port(), "3306"))
		c.DBName = strings.TrimPrefix(u.Path, "/")
		if u.User != nil {
			c.User = u.User.Username()
			c.Passwd, _ = u.User.Password()
		}
		db, err = sql.Open("mysql", c.FormatDSN())
	default:
		return errors.New("unsupported DATABASE_ENGINE")
	}
	if db != nil {
		db.SetMaxOpenConns(5)
	}
	return err
}
func envPort(v, fallback string) string {
	if v == "" {
		return fallback
	}
	return v
}
func migrate(ctx context.Context) error {
	if err := database(ctx); err != nil {
		return err
	}
	if engine == "mongodb" {
		_, err := collection.Indexes().CreateOne(ctx, mongo.IndexModel{Keys: bson.D{{Key: "id", Value: 1}}, Options: options.Index().SetUnique(true)})
		return err
	}
	_, err := db.ExecContext(ctx, "CREATE TABLE IF NOT EXISTS notes (id VARCHAR(36) PRIMARY KEY, body TEXT NOT NULL)")
	return err
}
func notes(ctx context.Context) ([]Note, error) {
	result := []Note{}
	if err := database(ctx); err != nil {
		return nil, err
	}
	if engine == "mongodb" {
		cursor, err := collection.Find(ctx, bson.D{}, options.Find().SetSort(bson.D{{Key: "id", Value: 1}}))
		if err != nil {
			return nil, err
		}
		defer cursor.Close(ctx)
		err = cursor.All(ctx, &result)
		return result, err
	}
	rows, err := db.QueryContext(ctx, "SELECT id, body FROM notes ORDER BY id")
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	for rows.Next() {
		var n Note
		if err = rows.Scan(&n.ID, &n.Body); err != nil {
			return nil, err
		}
		result = append(result, n)
	}
	return result, rows.Err()
}
func addNote(ctx context.Context, body string) (Note, error) {
	n := Note{ID: uuid.NewString(), Body: body}
	if err := database(ctx); err != nil {
		return n, err
	}
	if engine == "mongodb" {
		_, err := collection.InsertOne(ctx, n)
		return n, err
	}
	query := "INSERT INTO notes (id, body) VALUES ($1, $2)"
	if engine == "mariadb" {
		query = "INSERT INTO notes (id, body) VALUES (?, ?)"
	}
	_, err := db.ExecContext(ctx, query, n.ID, n.Body)
	return n, err
}

Worker

Creating an export enqueues an ID and returns immediately. This loop reads the notes, writes an export under that ID, records its status, and acknowledges the stream entry. It also reclaims jobs abandoned before acknowledgement. Retrying writes the same export key, so interruption does not create an additional export file; the snapshot can reflect newer notes if they changed before the retry. See worker recovery.

func worker(ctx context.Context) error {
	q, err := queue()
	if err != nil {
		return err
	}
	defer q.Close()
	err = q.XGroupCreateMkStream(ctx, "exports", "workers", "0").Err()
	if err != nil && !strings.Contains(err.Error(), "BUSYGROUP") {
		return err
	}
	consumer := uuid.NewString()
	log.Print("Worker ready")
	for ctx.Err() == nil {
		messages, _, err := q.XAutoClaim(ctx, &redis.XAutoClaimArgs{Stream: "exports", Group: "workers", Consumer: consumer, MinIdle: 30 * time.Second, Start: "0-0", Count: 1}).Result()
		if err == nil && len(messages) == 0 {
			streams, readErr := q.XReadGroup(ctx, &redis.XReadGroupArgs{Group: "workers", Consumer: consumer, Streams: []string{"exports", ">"}, Count: 1, Block: 5 * time.Second}).Result()
			err = readErr
			if len(streams) > 0 {
				messages = streams[0].Messages
			}
		}
		if errors.Is(err, redis.Nil) {
			continue
		}
		if err == nil {
			for _, message := range messages {
				err = process(ctx, q, message)
				if err != nil {
					break
				}
			}
		}
		if err != nil && ctx.Err() == nil {
			log.Print("Export failed; unacknowledged jobs retry after 30 seconds. Check attachments and schema.")
			time.Sleep(time.Second)
		}
	}
	return nil
}
func process(ctx context.Context, q *redis.Client, message redis.XMessage) error {
	id, _ := message.Values["id"].(string)
	if _, err := uuid.Parse(id); err != nil {
		return q.XAck(ctx, "exports", "workers", message.ID).Err()
	}
	state, err := q.Get(ctx, "export:"+id).Result()
	if err != nil && !errors.Is(err, redis.Nil) {
		return err
	}
	if state != "ready" {
		rows, err := notes(ctx)
		if err != nil {
			return err
		}
		data, err := json.MarshalIndent(rows, "", "  ")
		if err != nil {
			return err
		}
		if err = saveExport(ctx, q, id, data); err != nil {
			return err
		}
		if err = q.Set(ctx, "export:"+id, "ready", 0).Err(); err != nil {
			return err
		}
	}
	if err = q.XAck(ctx, "exports", "workers", message.ID).Err(); err != nil {
		return err
	}
	log.Printf("Export ready: %s", id)
	return q.XDel(ctx, "exports", message.ID).Err()
}

Object storage

Without S3 configuration, exports are stored in Redis for the worker checkpoint. With an attached bucket, the worker writes JSON to exports/<id>.json and the web process reads that object for downloads. Both processes need their own bucket attachment. The client uses the supplied endpoint and region with path-style requests. Credentials stay on the server; a private bucket does not add authentication to the example’s public download route.

func storage(ctx context.Context) (*s3.Client, error) {
	cfg, err := config.LoadDefaultConfig(ctx, config.WithRegion(os.Getenv("S3_REGION")))
	if err != nil {
		return nil, err
	}
	return s3.NewFromConfig(cfg, func(o *s3.Options) { o.BaseEndpoint = aws.String(os.Getenv("S3_ENDPOINT")); o.UsePathStyle = true }), nil
}
func saveExport(ctx context.Context, q *redis.Client, id string, data []byte) error {
	if os.Getenv("S3_BUCKET") == "" {
		return q.Set(ctx, "export:"+id+":data", data, 0).Err()
	}
	client, err := storage(ctx)
	if err != nil {
		return err
	}
	_, err = client.PutObject(ctx, &s3.PutObjectInput{Bucket: aws.String(os.Getenv("S3_BUCKET")), Key: aws.String("exports/" + id + ".json"), Body: bytes.NewReader(data), ContentType: aws.String("application/json")})
	return err
}
func loadExport(ctx context.Context, q *redis.Client, id string) ([]byte, error) {
	if os.Getenv("S3_BUCKET") == "" {
		return q.Get(ctx, "export:"+id+":data").Bytes()
	}
	client, err := storage(ctx)
	if err != nil {
		return nil, err
	}
	obj, err := client.GetObject(ctx, &s3.GetObjectInput{Bucket: aws.String(os.Getenv("S3_BUCKET")), Key: aws.String("exports/" + id + ".json")})
	if err != nil {
		return nil, err
	}
	defer obj.Body.Close()
	return io.ReadAll(obj.Body)
}

Complete application

The complete source and startup commands are in the notes ZIP linked above. The HTTP layer connects the three components: POST /notes validates and saves a record, GET /notes lists records, POST /exports queues work, and the export status and download routes expose the result. Follow the language’s tutorial sequence to exercise each part, then review production preparation.