Gluing the Real World: Go CLI for OSM → PostGIS → Tiles → Object Storage¶

package main

import (
    "context"
    "flag"
    "fmt"
    "log"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/spf13/cobra"
    "github.com/spf13/viper"
)

type Config struct {
    OSM struct {
        URL    string `mapstructure:"url"`
        SHA256 string `mapstructure:"sha256"`
    } `mapstructure:"osm"`

    DB struct {
        DSN string `mapstructure:"dsn"`
    } `mapstructure:"db"`

    Tiling struct {
        MinZoom int `mapstructure:"min_zoom"`
        MaxZoom int `mapstructure:"max_zoom"`
        Layers  []Layer `mapstructure:"layers"`
    } `mapstructure:"tiling"`

    Publish struct {
        S3Bucket string `mapstructure:"s3_bucket"`
        Prefix   string `mapstructure:"prefix"`
    } `mapstructure:"publish"`

    Metrics struct {
        Pushgateway string `mapstructure:"pushgateway"`
    } `mapstructure:"metrics"`
}

type Layer struct {
    Name string `mapstructure:"name"`
    SQL  string `mapstructure:"sql"`
}

var (
    cfg Config
    runID string
)

func main() {
    runID = fmt.Sprintf("osmglue-%d", time.Now().Unix())

    var rootCmd = &cobra.Command{
        Use:   "osmglue",
        Short: "OSM → PostGIS → Tiles pipeline orchestrator",
        Long:  "A Go CLI that orchestrates the complete OSM data pipeline",
    }

    rootCmd.AddCommand(bootCmd())
    rootCmd.AddCommand(fetchCmd())
    rootCmd.AddCommand(loadCmd())
    rootCmd.AddCommand(indexCmd())
    rootCmd.AddCommand(tileCmd())
    rootCmd.AddCommand(serveCmd())
    rootCmd.AddCommand(publishCmd())
    rootCmd.AddCommand(statusCmd())

    // Load config
    viper.SetConfigName("pipeline")
    viper.SetConfigType("yaml")
    viper.AddConfigPath(".")
    viper.AutomaticEnv()

    if err := viper.ReadInConfig(); err != nil {
        log.Fatal("Failed to read config:", err)
    }

    if err := viper.Unmarshal(&cfg); err != nil {
        log.Fatal("Failed to unmarshal config:", err)
    }

    if err := rootCmd.Execute(); err != nil {
        log.Fatal(err)
    }
}

func bootCmd() *cobra.Command {
    return &cobra.Command{
        Use:   "boot",
        Short: "Ensure docker services are running",
        Run: func(cmd *cobra.Command, args []string) {
            ctx := contextWithCancel()

            logJSON("boot_start", map[string]any{
                "run_id": runID,
                "services": []string{"postgis", "martin", "pushgateway"},
            })

            // Start docker compose
            if err := execWithRetry(ctx, Opts{
                Cmd:  "docker",
                Args: []string{"compose", "up", "-d"},
                Retries: 3,
            }); err != nil {
                log.Fatal("Failed to start services:", err)
            }

            // Wait for health checks
            time.Sleep(30 * time.Second)

            logJSON("boot_complete", map[string]any{
                "run_id": runID,
                "status": "healthy",
            })
        },
    }
}

func fetchCmd() *cobra.Command {
    return &cobra.Command{
        Use:   "fetch",
        Short: "Download OSM PBF and verify checksum",
        Run: func(cmd *cobra.Command, args []string) {
            ctx := contextWithCancel()

            logJSON("fetch_start", map[string]any{
                "run_id": runID,
                "url": cfg.OSM.URL,
            })

            start := time.Now()

            // Download PBF with resume support
            if err := execWithRetry(ctx, Opts{
                Cmd:  "curl",
                Args: []string{"-C", "-", "-o", "data.osm.pbf", cfg.OSM.URL},
                Retries: 3,
            }); err != nil {
                log.Fatal("Failed to download PBF:", err)
            }

            // Download checksum
            if err := execWithRetry(ctx, Opts{
                Cmd:  "curl",
                Args: []string{"-sS", "-o", "data.osm.pbf.sha256", cfg.OSM.SHA256},
                Retries: 3,
            }); err != nil {
                log.Fatal("Failed to download checksum:", err)
            }

            // Verify checksum
            if err := execWithRetry(ctx, Opts{
                Cmd:  "sha256sum",
                Args: []string{"-c", "data.osm.pbf.sha256"},
                Retries: 1,
            }); err != nil {
                log.Fatal("Checksum verification failed:", err)
            }

            duration := time.Since(start)

            logJSON("fetch_complete", map[string]any{
                "run_id": runID,
                "duration_seconds": duration.Seconds(),
                "status": "verified",
            })

            // Push metrics
            pushMetrics("fetch", duration.Seconds())
        },
    }
}

func loadCmd() *cobra.Command {
    return &cobra.Command{
        Use:   "load",
        Short: "Load OSM data into PostGIS via osm2pgsql",
        Run: func(cmd *cobra.Command, args []string) {
            ctx := contextWithCancel()

            logJSON("load_start", map[string]any{
                "run_id": runID,
                "pbf_path": "data.osm.pbf",
            })

            start := time.Now()

            // Run osm2pgsql in container
            if err := execWithRetry(ctx, Opts{
                Cmd: "docker",
                Args: []string{
                    "run", "--rm", "--network=host",
                    "-v", fmt.Sprintf("%s:/data.osm.pbf", "data.osm.pbf"),
                    "openmaptiles/osm2pgsql",
                    "--create", "--slim",
                    "-H", "localhost", "-d", "osm", "-U", "postgres",
                    "--cache", "20000", "--number-processes", "4",
                    "/data.osm.pbf",
                },
                Retries: 1,
            }); err != nil {
                log.Fatal("Failed to load OSM data:", err)
            }

            duration := time.Since(start)

            logJSON("load_complete", map[string]any{
                "run_id": runID,
                "duration_seconds": duration.Seconds(),
                "status": "loaded",
            })

            pushMetrics("load", duration.Seconds())
        },
    }
}

func indexCmd() *cobra.Command {
    return &cobra.Command{
        Use:   "index",
        Short: "Create indexes and materialized views",
        Run: func(cmd *cobra.Command, args []string) {
            ctx := contextWithCancel()

            logJSON("index_start", map[string]any{
                "run_id": runID,
            })

            start := time.Now()

            // Execute SQL indexing script
            sqlScript := `
                -- Add spatial indexes
                CREATE INDEX IF NOT EXISTS idx_roads_geom ON planet_osm_roads USING gist(way);
                CREATE INDEX IF NOT EXISTS idx_line_geom ON planet_osm_line USING gist(way);
                CREATE INDEX IF NOT EXISTS idx_polygon_geom ON planet_osm_polygon USING gist(way);

                -- Analyze tables
                ANALYZE planet_osm_roads;
                ANALYZE planet_osm_line;
                ANALYZE planet_osm_polygon;

                -- Create materialized views
                CREATE MATERIALIZED VIEW IF NOT EXISTS roads_z8 AS
                SELECT osm_id, highway, name,
                       ST_SimplifyVW(ST_Transform(way, 3857), 30) AS geom
                FROM planet_osm_roads
                WHERE highway IS NOT NULL;

                CREATE INDEX IF NOT EXISTS idx_roads_z8_geom ON roads_z8 USING gist(geom);

                REFRESH MATERIALIZED VIEW roads_z8;
            `

            if err := execWithRetry(ctx, Opts{
                Cmd:  "psql",
                Args: []string{"-d", cfg.DB.DSN, "-c", sqlScript},
                Retries: 3,
            }); err != nil {
                log.Fatal("Failed to create indexes:", err)
            }

            duration := time.Since(start)

            logJSON("index_complete", map[string]any{
                "run_id": runID,
                "duration_seconds": duration.Seconds(),
                "status": "indexed",
            })

            pushMetrics("index", duration.Seconds())
        },
    }
}

func tileCmd() *cobra.Command {
    return &cobra.Command{
        Use:   "tile",
        Short: "Generate tiles with tippecanoe",
        Run: func(cmd *cobra.Command, args []string) {
            ctx := contextWithCancel()

            logJSON("tile_start", map[string]any{
                "run_id": runID,
                "min_zoom": cfg.Tiling.MinZoom,
                "max_zoom": cfg.Tiling.MaxZoom,
            })

            start := time.Now()

            // Build tippecanoe command with layers
            args := []string{
                "-o", "tiles.mbtiles",
                "-Z", fmt.Sprintf("%d", cfg.Tiling.MinZoom),
                "-z", fmt.Sprintf("%d", cfg.Tiling.MaxZoom),
                "--drop-densest-as-needed",
                "--extend-zooms-if-still-dropping",
                "--no-feature-limit",
                "--no-tile-size-limit",
                "--read-parallel",
            }

            // Add layers
            for _, layer := range cfg.Tiling.Layers {
                args = append(args, "-L", fmt.Sprintf("name=%s,sql=%s", layer.Name, layer.SQL))
            }

            if err := execWithRetry(ctx, Opts{
                Cmd:  "tippecanoe",
                Args: args,
                Retries: 2,
            }); err != nil {
                log.Fatal("Failed to generate tiles:", err)
            }

            duration := time.Since(start)

            logJSON("tile_complete", map[string]any{
                "run_id": runID,
                "duration_seconds": duration.Seconds(),
                "output": "tiles.mbtiles",
                "status": "generated",
            })

            pushMetrics("tile", duration.Seconds())
        },
    }
}

func publishCmd() *cobra.Command {
    return &cobra.Command{
        Use:   "publish",
        Short: "Publish tiles to S3/MinIO",
        Run: func(cmd *cobra.Command, args []string) {
            ctx := contextWithCancel()

            logJSON("publish_start", map[string]any{
                "run_id": runID,
                "bucket": cfg.Publish.S3Bucket,
                "prefix": cfg.Publish.Prefix,
            })

            start := time.Now()

            // Use rclone to copy to S3/MinIO
            if err := execWithRetry(ctx, Opts{
                Cmd:  "rclone",
                Args: []string{"copy", "tiles.mbtiles", fmt.Sprintf("s3:%s/%s", cfg.Publish.S3Bucket, cfg.Publish.Prefix)},
                Retries: 3,
            }); err != nil {
                log.Fatal("Failed to publish tiles:", err)
            }

            duration := time.Since(start)

            logJSON("publish_complete", map[string]any{
                "run_id": runID,
                "duration_seconds": duration.Seconds(),
                "status": "published",
            })

            pushMetrics("publish", duration.Seconds())
        },
    }
}

// cmd/internal/run/run.go
package run

import (
    "context"
    "encoding/json"
    "log"
    "math/rand"
    "os/exec"
    "time"
)

type Opts struct {
    Cmd     string
    Args    []string
    Env     []string
    Dir     string
    Retries int
}

func ExecWithRetry(ctx context.Context, o Opts) error {
    delay := 1 * time.Second
    maxDelay := 30 * time.Second

    for attempt := 0; attempt <= o.Retries; attempt++ {
        cmd := exec.CommandContext(ctx, o.Cmd, o.Args...)
        cmd.Dir = o.Dir
        cmd.Env = append(cmd.Env, o.Env...)

        out, err := cmd.CombinedOutput()

        logJSON("exec", map[string]any{
            "cmd":     o.Cmd,
            "args":    o.Args,
            "attempt": attempt,
            "out":     string(out),
            "err":     err,
        })

        if err == nil {
            return nil
        }

        if attempt == o.Retries {
            return err
        }

        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-time.After(delay + time.Duration(rand.Intn(500))*time.Millisecond):
            delay *= 2
            if delay > maxDelay {
                delay = maxDelay
            }
        }
    }

    return context.DeadlineExceeded
}

func logJSON(ev string, payload map[string]any) {
    payload["event"] = ev
    payload["timestamp"] = time.Now().UTC().Format(time.RFC3339)
    payload["run_id"] = runID

    b, _ := json.Marshal(payload)
    log.Println(string(b))
}

func contextWithCancel() context.Context {
    ctx, cancel := context.WithCancel(context.Background())

    c := make(chan os.Signal, 1)
    signal.Notify(c, os.Interrupt, syscall.SIGTERM)

    go func() {
        <-c
        logJSON("shutdown", map[string]any{"signal": "interrupt"})
        cancel()
    }()

    return ctx
}

func pushMetrics(stage string, duration float64) {
    if cfg.Metrics.Pushgateway == "" {
        return
    }

    metrics := fmt.Sprintf(`
# TYPE stage_duration_seconds gauge
stage_duration_seconds{stage="%s",run_id="%s"} %f
`, stage, runID, duration)

    execWithRetry(context.Background(), Opts{
        Cmd: "curl",
        Args: []string{
            "-X", "POST",
            "--data-binary", metrics,
            fmt.Sprintf("%s/metrics/job/%s", cfg.Metrics.Pushgateway, stage),
        },
        Retries: 1,
    })
}

// Success marker files per stage
func markStageComplete(stage string) error {
    return os.WriteFile(fmt.Sprintf(".%s.ok", stage), []byte(runID), 0644)
}

func isStageComplete(stage string) bool {
    _, err := os.Stat(fmt.Sprintf(".%s.ok", stage))
    return err == nil
}

// Skip stages unless --force flag
func shouldSkipStage(stage string, force bool) bool {
    return !force && isStageComplete(stage)
}

// Write to temp file then move into place
func atomicWrite(filename string, data []byte) error {
    tempFile := filename + ".tmp"
    if err := os.WriteFile(tempFile, data, 0644); err != nil {
        return err
    }
    return os.Rename(tempFile, filename)
}

func waitForPostGIS(ctx context.Context, dsn string) error {
    for {
        select {
        case <-ctx.Done():
            return ctx.Err()
        default:
            db, err := sql.Open("postgres", dsn)
            if err == nil {
                if err := db.Ping(); err == nil {
                    db.Close()
                    return nil
                }
                db.Close()
            }
            time.Sleep(5 * time.Second)
        }
    }
}

func waitForMartin(ctx context.Context, url string) error {
    for {
        select {
        case <-ctx.Done():
            return ctx.Err()
        default:
            resp, err := http.Get(url + "/health")
            if err == nil && resp.StatusCode == 200 {
                resp.Body.Close()
                return nil
            }
            time.Sleep(5 * time.Second)
        }
    }
}

// Bound parallelism to CPU cores - 1
func getOptimalParallelism() int {
    cores := runtime.NumCPU()
    if cores > 1 {
        return cores - 1
    }
    return 1
}

// Use in tippecanoe
args := append(args, "--read-parallel", fmt.Sprintf("%d", getOptimalParallelism()))

type StageMetrics struct {
    Stage     string    `json:"stage"`
    StartTime time.Time `json:"start_time"`
    EndTime   time.Time `json:"end_time"`
    Duration  float64   `json:"duration_seconds"`
    Bytes     int64     `json:"bytes_processed"`
    Tiles     int       `json:"tiles_generated"`
    Status    string    `json:"status"`
}

func recordStageMetrics(stage string, start time.Time, bytes int64, tiles int) {
    duration := time.Since(start).Seconds()

    metrics := StageMetrics{
        Stage:     stage,
        StartTime: start,
        EndTime:   time.Now(),
        Duration:  duration,
        Bytes:     bytes,
        Tiles:     tiles,
        Status:    "success",
    }

    logJSON("stage_complete", map[string]any{
        "metrics": metrics,
    })

    pushMetrics(stage, duration)
}

// Cache downloads with checksum gates
func shouldRedownload(pbfPath, sha256Path string) bool {
    if _, err := os.Stat(pbfPath); os.IsNotExist(err) {
        return true
    }

    if _, err := os.Stat(sha256Path); os.IsNotExist(err) {
        return true
    }

    // Check if checksum matches
    cmd := exec.Command("sha256sum", "-c", sha256Path)
    return cmd.Run() != nil
}

// Reuse MBTiles when input unchanged
func shouldRegenerateTiles(mbtilesPath, pbfPath string) bool {
    if _, err := os.Stat(mbtilesPath); os.IsNotExist(err) {
        return true
    }

    // Compare modification times
    mbtilesInfo, _ := os.Stat(mbtilesPath)
    pbfInfo, _ := os.Stat(pbfPath)

    return mbtilesInfo.ModTime().Before(pbfInfo.ModTime())
}

// Keep last N artifacts with versioned prefixes
func publishWithVersion(prefix string) string {
    version := time.Now().Format("200601021504")
    return fmt.Sprintf("%s/v%s/", prefix, version)
}

// Cleanup old versions (keep last 5)
func cleanupOldVersions(bucket, prefix string) error {
    // Implementation depends on your storage backend
    // Use rclone or AWS CLI to list and delete old versions
    return nil
}