|
| 1 | +# versionedjob [](https://github.com/riverqueue/rivercontrib/actions) [](https://pkg.go.dev/github.com/riverqueue/rivercontrib/versionedjob) |
| 2 | + |
| 3 | +Provides a River hook with a simple job versioning framework. **Version transformers** are written for versioned jobs containing procedures for upgrading jobs that were encoded as older versions to the most modern version. This allows for workers to be implemented as if all job versions will be the most modern version only, keeping code simpler. |
| 4 | + |
| 5 | +```go |
| 6 | +// VersionTransformer defines how to perform transformations between versions |
| 7 | +// for a specific job kind. |
| 8 | +type VersionTransformer interface { |
| 9 | + // Kind is the job kind that this transformer applies to. |
| 10 | + Kind() string |
| 11 | + |
| 12 | + // VersionTransform applies version transformations to the given job. Version |
| 13 | + // transformations are fully defined according to the user, as well as how a |
| 14 | + // version is extracted from the job's args. |
| 15 | + // |
| 16 | + // Generally, this function should extract a version from the job, then |
| 17 | + // apply versions one by one until it's fully modernized to the point where |
| 18 | + // it can be successfully run by its worker. |
| 19 | + VersionTransform(job *rivertype.JobRow) error |
| 20 | +} |
| 21 | +``` |
| 22 | + |
| 23 | +## Example |
| 24 | + |
| 25 | +Below are three versions of the same job: `VersionedJobArgsV1`, `VersionedJobArgsV2`, and the current version, `VersionedJobArgs`. From V1 to V2, `name` was renamed to `title`, and a `version` field added to track version. In V3, a new `description` property was added. A real program would only keep the latest version (`VersionedJobArgs`), but this example shows all three for reference. |
| 26 | + |
| 27 | +```go |
| 28 | +type VersionedJobArgsV1 struct { |
| 29 | + Name string `json:"name"` |
| 30 | +} |
| 31 | + |
| 32 | +type VersionedJobArgsV2 struct { |
| 33 | + Title string `json:"title"` |
| 34 | + Version int `json:"version"` |
| 35 | +} |
| 36 | + |
| 37 | +type VersionedJobArgs struct { |
| 38 | + Description string `json:"description"` |
| 39 | + Title string `json:"title"` |
| 40 | + Version int `json:"version"` |
| 41 | +} |
| 42 | +``` |
| 43 | + |
| 44 | +The worker for `VersionedJobArgs` is written so it only handles the latest version (`title` instead of `name` and assumes `description` is present). This is possible because a `VersionTransformer` will handle migrating jobs from old versions to new ones before they hit the worker. |
| 45 | + |
| 46 | +```go |
| 47 | +type VersionedJobWorker struct { |
| 48 | + river.WorkerDefaults[VersionedJobArgs] |
| 49 | +} |
| 50 | + |
| 51 | +func (w *VersionedJobWorker) Work(ctx context.Context, job *river.Job[VersionedJobArgs]) error { |
| 52 | + fmt.Printf("Job title: %s; description: %s\n", job.Args.Title, job.Args.Description) |
| 53 | + return nil |
| 54 | +} |
| 55 | +``` |
| 56 | + |
| 57 | +The `VersionTransformer` implementation handles version upgrades one by one. Jobs which are multiple versions old can still be upgraded because multiple version changes can be applied in one go. This implementation uses `gjson`/`sjson` so that each change need only know a minimum about the data object in question and that unknown fields are retained. Other approaches are possible though, including using only Go's built-in `gjson` package. |
| 58 | + |
| 59 | +```go |
| 60 | +type VersionedJobTransformer struct{} |
| 61 | + |
| 62 | +func (*VersionedJobTransformer) VersionTransform(ctx context.Context, job *rivertype.JobRow) error { |
| 63 | + // Extract version from job, defaulting to 1 if not present because we |
| 64 | + // assume that was before versioning was introduced. |
| 65 | + version := cmp.Or(gjson.GetBytes(job.EncodedArgs, "version").Int(), 1) |
| 66 | + |
| 67 | + var err error |
| 68 | + |
| 69 | + // |
| 70 | + // Here, we walk through each successive version, applying transformations |
| 71 | + // to bring it to its next version. If a job is multiple versions behind, |
| 72 | + // version transformations are one-by-one applied in order until the job's |
| 73 | + // args are fully modernized. |
| 74 | + // |
| 75 | + |
| 76 | + // Version change: V1 --> V2 |
| 77 | + if version < 2 { |
| 78 | + version = 2 |
| 79 | + |
| 80 | + job.EncodedArgs, err = sjson.SetBytes(job.EncodedArgs, "title", gjson.GetBytes(job.EncodedArgs, "name").String()) |
| 81 | + if err != nil { |
| 82 | + return err |
| 83 | + } |
| 84 | + |
| 85 | + job.EncodedArgs, err = sjson.DeleteBytes(job.EncodedArgs, "name") |
| 86 | + if err != nil { |
| 87 | + return err |
| 88 | + } |
| 89 | + } |
| 90 | + |
| 91 | + // Version change: V2 --> V3 |
| 92 | + if version < 3 { |
| 93 | + version = 3 |
| 94 | + |
| 95 | + title := gjson.GetBytes(job.EncodedArgs, "title").String() |
| 96 | + if title == "" { |
| 97 | + return errors.New("no title found in job args") |
| 98 | + } |
| 99 | + |
| 100 | + job.EncodedArgs, err = sjson.SetBytes(job.EncodedArgs, "description", "A description of a "+title+".") |
| 101 | + if err != nil { |
| 102 | + return err |
| 103 | + } |
| 104 | + } |
| 105 | + |
| 106 | + // Not strictly necessary, but set version to latest. |
| 107 | + job.EncodedArgs, err = sjson.SetBytes(job.EncodedArgs, "version", version) |
| 108 | + if err != nil { |
| 109 | + return err |
| 110 | + } |
| 111 | + |
| 112 | + return nil |
| 113 | +} |
| 114 | + |
| 115 | +func (*VersionedJobTransformer) Kind() string { return (VersionedJobArgs{}).Kind() } |
| 116 | +``` |
| 117 | + |
| 118 | +A River client is initialized with the `versiondjob` hook and transformer installed: |
| 119 | + |
| 120 | +```go |
| 121 | +riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{ |
| 122 | + Hooks: []rivertype.Hook{ |
| 123 | + versionedjob.NewHook(&versionedjob.HookConfig{ |
| 124 | + Transformers: []versionedjob.VersionTransformer{ |
| 125 | + &VersionedJobTransformer{}, |
| 126 | + }, |
| 127 | + }), |
| 128 | + }, |
| 129 | +}) |
| 130 | +if err != nil { |
| 131 | + panic(err) |
| 132 | +} |
| 133 | +``` |
| 134 | + |
| 135 | +With all that in place, a job of any version can be inserted and thanks to the version transformer modernizing the older ones, the worker will produce the same result regardless of input. |
| 136 | + |
| 137 | +```go |
| 138 | +if _, err = riverClient.InsertMany(ctx, []river.InsertManyParams{ |
| 139 | + { |
| 140 | + Args: VersionedJobArgsV1{ |
| 141 | + Name: "My Job", |
| 142 | + }, |
| 143 | + }, |
| 144 | + { |
| 145 | + Args: VersionedJobArgsV2{ |
| 146 | + Title: "My Job", |
| 147 | + Version: 2, |
| 148 | + }, |
| 149 | + }, |
| 150 | + { |
| 151 | + Args: VersionedJobArgs{ |
| 152 | + Title: "My Job", |
| 153 | + Description: "A description of a My Job.", |
| 154 | + Version: 3, |
| 155 | + }, |
| 156 | + }, |
| 157 | +}); err != nil { |
| 158 | + panic(err) |
| 159 | +} |
| 160 | +``` |
| 161 | + |
| 162 | +```go |
| 163 | +// Output: |
| 164 | +// Job title: My Job; description: A description of a My Job. |
| 165 | +// Job title: My Job; description: A description of a My Job. |
| 166 | +// Job title: My Job; description: A description of a My Job. |
| 167 | +``` |
0 commit comments