distributor/docs/consumers/api.md

# Upstream Producer Integration

Audience: developers and LLM coding agents adding `distributor` support to an upstream Go producer application.

This document is the copyable implementation guide for submitting producer outputs to a `distributor` pipeline whose source backend is `http_upload`.

## Required Inputs

The upstream application needs these values from deployment or operator configuration:

- distributor endpoint: the HTTP server base URL, such as `https://distributor.example.com`;
- upload token: bearer token that authenticates the producer;
- pipeline id: configured `http_upload` pipeline that should process this upload;
- generated files: regular local files to include in the source bundle;
- bundle id: stable identifier for the logical report stream or artifact;
- idempotency key: unique key for one producer run, reused only when retrying that same run.

Do not put destination routing, public URLs, transform settings, or credentials in the source manifest. Those belong in the `distributor` pipeline configuration.

The token, pipeline id, bundle id, and idempotency key have different jobs. The token authenticates the producer. The pipeline id selects the configured distributor workflow, including destinations and publishing policy. The bundle id tells `distributor` whether a new upload is a newer version of the same source; keep it stable across runs that should replace the same managed destination artifact. The idempotency key tells `distributor` whether an upload request is a retry; change it for each distinct producer run so new content is enqueued.

## Recommended Workflow

Use `gitea.maximumdirect.net/eric/distributor/pkg/upload`.

For most producers, use `UploadFiles`. It accepts producer-generated files, builds a temporary valid source bundle with `pkg/bundle`, uploads a gzip-compressed tar archive, and removes temporary files when the call returns.

Use `UploadBundle` only when the producer already assembled a complete bundle directory containing `manifest.json`.

Add the dependency from the upstream application:

```sh
go get gitea.maximumdirect.net/eric/distributor
```

## Minimal Go Example

```go
package reports

import (
	"context"
	"errors"
	"fmt"
	"os"
	"time"

	"gitea.maximumdirect.net/eric/distributor/pkg/bundle"
	"gitea.maximumdirect.net/eric/distributor/pkg/upload"
)

func SubmitReport(reportPath, summaryPath string) error {
	endpoint := os.Getenv("DISTRIBUTOR_UPLOAD_ENDPOINT")
	token := os.Getenv("DISTRIBUTOR_UPLOAD_TOKEN")
	if endpoint == "" || token == "" {
		return fmt.Errorf("distributor endpoint and token are required")
	}

	pipelineID := "weather-hourly"
	reportID := "weather.hourly.brentwood"
	runID := time.Now().UTC().Format("20060102T150405.000000000Z")
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	client, err := upload.NewClient(upload.ClientOptions{
		Endpoint: endpoint,
		Token:    token,
	})
	if err != nil {
		return err
	}

	result, err := client.UploadFiles(ctx, upload.UploadFilesOptions{
		PipelineID:     pipelineID,
		ID:             reportID,
		IdempotencyKey: reportID + "." + runID,
		Files: []bundle.BundleFile{
			{SourcePath: reportPath, Path: "report.md"},
			{SourcePath: summaryPath, Path: "summary.txt"},
		},
	})
	if err != nil {
		var conflict *upload.IdempotencyConflictError
		if errors.As(err, &conflict) {
			return fmt.Errorf("idempotency key was reused for different bundle content: %w", err)
		}
		return err
	}

	fmt.Printf("distributor accepted run %s\n", result.RunID)
	return nil
}
```

## Producer Responsibilities

- Use a stable bundle id for the logical producer output that should replace the same destination artifact, such as `weather.hourly.brentwood`.
- Set `PipelineID` to the configured upload pipeline that should process the bundle.
- Do not include per-run timestamps, random values, or job ids in the bundle id unless each run should be treated as a different source.
- Use an idempotency key that changes for every distinct producer run, such as `<bundle-id>.<run-id>`.
- Reuse the same idempotency key only when retrying the exact same producer run with the same source manifest.
- Map each generated file to a clean slash-separated bundle path, such as `report.md` or `assets/chart.png`.
- Include only regular files. Symlinks, directories as files, devices, FIFOs, and sockets are rejected.
- Keep file contents stable after upload inputs are selected. Bundle digests are calculated from file bytes.
- Treat upload success as admission only. `UploadFiles` and `UploadBundle` return after the server accepts and validates the upload, not after all destinations publish.

Valid bundle paths are relative slash paths. They must not be empty, absolute, contain backslashes, contain `.` or `..` path segments, contain empty path segments, or use reserved basenames `manifest.json` or `.distributor.json`.

## Idempotency And Status

`pkg/upload` sends `Idempotency-Key` on every upload. If the caller omits one, the package generates a random key for that call and reuses it for in-process retries. That is enough for transient network retry within one process, but it does not give cross-process retry identity.

For producer jobs that may retry after process restart, supply a key derived from the producer run, such as `<bundle-id>.<run-id>`. Reusing the same key with the same token, pipeline id, and normalized source manifest returns the original accepted run. Reusing the same key with different source content in that scope returns a conflict. Reusing one key across multiple distinct report generations prevents those generations from being treated as new uploads.

`Status` polls `/runs/<run-id>` while the distributor server retains the in-memory status record. Status values are `accepted`, `queued`, `running`, `succeeded`, and `failed`. Completed records expire according to the server's `server.http.retention` setting, and server restart clears status and idempotency records.

Optional status check:

```go
status, err := client.Status(ctx, result.RunID)
if err != nil {
	return err
}
if status.Status == "failed" {
	return fmt.Errorf("distributor run failed: %s", status.Error)
}
```

## References

In the `distributor` source tree:

- `docs/consumers/pkg-upload.md`: Go upload package workflow.
- `docs/consumers/pkg-bundle.md`: Go bundle package workflow.
- `docs/integrations/http-upload.md`: canonical HTTP upload wire contract.
- `docs/integrations/source-bundle.md`: canonical source bundle file-format contract.