lode - v0.1.0

Constraints and Upserts

In this guide we will learn how to use constraints and upserts. To make the discussion concrete we will build a tagging system for blog posts: a post may have many tags, and a tag belongs to many posts. This is a classic many-to-many relationship, and it hides a classic concurrency problem — two requests creating the tag "gleam" at the same time — that constraints and upserts exist to solve.

Our storage needs three tables, declared as a migration. Migrations in lode are ordinary values built with the lode/migration module from the lode_sql package (see Getting Started for running them with the migrator):

import lode/migration
import lode/migration/ddl

pub fn add_posts_and_tags() -> migration.Migration {
  migration.new(20_260_612_120_000)
  |> migration.create_table("posts", [
    ddl.column("id", ddl.Serial) |> ddl.primary_key,
    ddl.column("title", ddl.Text) |> ddl.not_null,
    ddl.column("body", ddl.Text) |> ddl.not_null,
  ])
  |> migration.create_table("tags", [
    ddl.column("id", ddl.Serial) |> ddl.primary_key,
    ddl.column("name", ddl.Text) |> ddl.not_null,
  ])
  |> migration.create_index("tags", ["name"], unique: True)
  |> migration.create_table("posts_tags", [
    ddl.column("post_id", ddl.Integer)
      |> ddl.not_null
      |> ddl.references(ddl.reference("posts")),
    ddl.column("tag_id", ddl.Integer)
      |> ddl.not_null
      |> ddl.references(ddl.reference("tags")),
  ])
}

Note the unique index on tags.name. We could try to enforce uniqueness in application code — look the tag up, insert it only if missing — but between the lookup and the insert another request may insert the same tag. Validations are useful for user feedback; only the database can guarantee uniqueness. The index created by create_index("tags", ["name"], unique: True) gets the conventional name tags_name_index, which we will refer back to shortly.

Note — DDL builders. ddl.references(column, ddl.reference("posts")) renders the foreign key inline (pipe ddl.on_delete/ddl.on_update for referential actions), and ..ddl.timestamps() splices the conventional inserted_at/updated_at columns into a table. Genuinely unsupported DDL (e.g. exclusion constraints) still uses the raw-SQL migration.execute(up:, down:) escape hatch. See the DDL coverage notes.

The schemas are plain values (no macros — see DESIGN.md in the repository), with the many-to-many association registered on the post:

import lode/association
import lode/schema.{type Schema}
import lode/types/primitive
import lode/value.{VInt, VString}
import gleam/dict
import gleam/result

pub type Tag {
  Tag(id: Int, name: String)
}

pub type Post {
  Post(id: Int, title: String, body: String, tags: List(Tag))
}

pub fn tag_schema() -> Schema(Tag) {
  schema.new(
    source: "tags",
    fields: [
      schema.primary_key(name: "id", of: primitive.id()),
      schema.field(name: "name", of: primitive.string()),
    ],
    load: fn(row) {
      use id <- result.try(schema.require_int(row, "id"))
      use name <- result.try(schema.require_string(row, "name"))
      Ok(Tag(id:, name:))
    },
    dump: fn(t: Tag) {
      dict.from_list([#("id", VInt(t.id)), #("name", VString(t.name))])
    },
  )
}

pub fn post_schema() -> Schema(Post) {
  schema.new(
    source: "posts",
    fields: [
      schema.primary_key(name: "id", of: primitive.id()),
      schema.field(name: "title", of: primitive.string()),
      schema.field(name: "body", of: primitive.string()),
    ],
    load: fn(row) {
      use id <- result.try(schema.require_int(row, "id"))
      use title <- result.try(schema.require_string(row, "title"))
      use body <- result.try(schema.require_string(row, "body"))
      Ok(Post(id:, title:, body:, tags: []))
    },
    dump: fn(p: Post) {
      dict.from_list([
        #("id", VInt(p.id)),
        #("title", VString(p.title)),
        #("body", VString(p.body)),
      ])
    },
  )
  |> association.many_to_many(
    name: "tags",
    related: tag_schema,
    join_through: "posts_tags",
    join_owner_key: "post_id",
    join_related_key: "tag_id",
    owner_key: fn(p: Post) { VInt(p.id) },
    child_key: fn(t: Tag) { VInt(t.id) },
    put: fn(p, tags) { Post(..p, tags: tags) },
    opts: association.options()
      |> association.on_replace(association.ReplaceDelete),
  )
}

The on_replace(ReplaceDelete) option says that when we write a new tag list for a post, any join rows for tags missing from the new list are deleted (the tags themselves survive — only the link is removed).

put_assoc vs cast_assoc

Imagine the user submits tags as a comma-separated string, "gleam, beam". changeset.cast_assoc is the wrong tool here: it builds child changesets from structured parameter maps, and what we have is a string that needs parsing and a get-or-insert step per tag. changeset.put_assoc is the right tool — we process the input ourselves and hand the changeset finished child changesets:

import lode/changeset
import lode/error.{type LodeError}
import lode/repo.{type Repo}
import gleam/list
import gleam/option.{None, Some}
import gleam/string

fn parse_tag_names(input: String) -> List(String) {
  input
  |> string.split(",")
  |> list.map(string.trim)
  |> list.filter(fn(name) { name != "" })
}

pub fn create_post(
  r: Repo,
  title title: String,
  body body: String,
  tags tags: String,
) -> Result(Post, LodeError) {
  use tag_rows <- result.try(
    parse_tag_names(tags)
    |> list.try_map(fn(name) { get_or_insert_tag(r, name) }),
  )
  let cs =
    changeset.change(Post(id: 0, title: "", body: "", tags: []), post_schema())
    |> changeset.put_change(field.field("title"), VString(title))
    |> changeset.put_change(field.field("body"), VString(body))
    |> changeset.put_assoc("tags", list.map(tag_rows, fn(tag) {
      changeset.change(tag, tag_schema())
    }))
  repo.insert(r, post_schema(), cs)
}

fn get_or_insert_tag(r: Repo, name: String) -> Result(Tag, LodeError) {
  use existing <- result.try(
    repo.get_by(r, tag_schema(), [#("name", VString(name))]),
  )
  case existing {
    Some(tag) -> Ok(tag)
    None -> repo.insert(r, tag_schema(), new_tag(name))
  }
}

fn new_tag(name: String) -> changeset.Changeset(Tag) {
  changeset.cast(
    Tag(id: 0, name: ""),
    tag_schema(),
    dict.from_list([#("name", VString(name))]),
    field.fields(["name"]),
  )
  |> changeset.validate_required(field.fields(["name"]))
}

Two notes on the translation from Ecto. First, Ecto’s version calls the repo from inside the schema module; lode has no global repo (an intentional drop — see DESIGN.md in the repository), so the repo is an explicit argument. Second, the tags we pass to put_assoc are changesets over rows that already carry their primary keys, so the association write doesn’t try to re-insert them — it only reconciles the posts_tags join rows, honoring on_replace.

This works, but get_or_insert_tag has a hole.

Constraints and race conditions

Between repo.get_by finding no tag and repo.insert creating it, a concurrent request may insert the same tag. Both lookups miss, both insert, and the unique index rejects one of them. The race is rare per request and near-certain over the life of an application.

In Elixir’s Ecto this failure is an Ecto.ConstraintError exception unless you declare unique_constraint/2 on the changeset, which converts it into a {:error, changeset} tuple. lode never raises: every repo call already returns a Result, and a violated constraint comes back as the ConstraintError variant of LodeError, carrying the constraint’s kind and database name. So recovering from the race is ordinary pattern matching:

import lode/error

fn get_or_insert_tag(r: Repo, name: String) -> Result(Tag, LodeError) {
  use existing <- result.try(
    repo.get_by(r, tag_schema(), [#("name", VString(name))]),
  )
  case existing {
    Some(tag) -> Ok(tag)
    None -> insert_or_fetch_tag(r, name)
  }
}

fn insert_or_fetch_tag(r: Repo, name: String) -> Result(Tag, LodeError) {
  case repo.insert(r, tag_schema(), new_tag(name)) {
    Ok(tag) -> Ok(tag)
    // Lost the race: someone inserted this tag after our lookup missed.
    Error(error.ConstraintError(kind: error.Unique, ..)) -> {
      use found <- result.try(
        repo.get_by(r, tag_schema(), [#("name", VString(name))]),
      )
      option.to_result(found, error.NoResults)
    }
    Error(other) -> Error(other)
  }
}

Like Ecto’s optimized version, this costs one query per existing tag, two queries per new tag, and three in the rare case the race actually happens — and it can no longer corrupt or duplicate data, because the database has the final word.

The changeset side of Ecto’s mechanism exists too: changeset.unique_constraint(field.field("name"), "tags_name_index") records the constraint on the changeset (note that the database constraint name is explicit — without reflection, lode cannot infer tags_name_index from the field). foreign_key_constraint and check_constraint work the same way.

When a constraint is declared on the changeset, a violation whose database name matches it is converted into a changeset field error: repo.insert/repo.update return Error(error.ChangesetInvalid(errors)) with the constraint’s message on the field ("has already been taken"), ready for form rendering — exactly like Ecto’s {:error, changeset}. A violation with no matching declared constraint still surfaces as the raw Error(error.ConstraintError(..)), so the race-recovery pattern above (matching ConstraintError directly) remains available when you’d rather handle it yourself.

Note — exclusion and association constraints. changeset.exclusion_constraint and changeset.no_assoc_constraint map those violations to field errors, just like unique_constraint/foreign_key_constraint. The exclusion constraint itself is still created with migration.execute. TASK:exclusion-constraint

Note — join-table constraints. A constraint on a many_to_many join table has no changeset to declare it on; declare it on the association instead: association.join_constraint(name:, message:) maps the violation to a field error on the association and rolls the write back (see the Associations cheatsheet).

Upserts

The pattern above asks the database for forgiveness; upserts ask it for cooperation. PostgreSQL 9.5+ supports INSERT ... ON CONFLICT, which lets the insert itself say what should happen when a constraint would be violated. Where Ecto threads :on_conflict/:conflict_target options through Repo.insert, lode gives the upserting insert its own function, repo.upsert, taking a policy from the lode/on_conflict module.

The simplest policy is do nothing — if the tag already exists, skip the insert entirely:

import lode/on_conflict

fn get_or_insert_tag(r: Repo, name: String) -> Result(Tag, LodeError) {
  use inserted <- result.try(repo.upsert(
    repo: r,
    schema: tag_schema(),
    changeset: new_tag(name),
    on_conflict: on_conflict.Nothing(target: on_conflict.Columns([field.field("name")])),
  ))
  case inserted {
    Some(tag) -> Ok(tag)
    // The insert was skipped: the tag already exists, so read it back.
    None -> {
      use found <- result.try(
        repo.get_by(r, tag_schema(), [#("name", VString(name))]),
      )
      option.to_result(found, error.NoResults)
    }
  }
}

In Ecto, on_conflict: :nothing on a conflicting insert hands back a struct without its primary key — a wart the guide warns about. lode makes the skip explicit instead: repo.upsert returns Ok(None) when a Nothing policy skipped the insert, so there is no half-real struct to misuse, but you do need the read-back query shown above to get the existing row.

If you want the row back in one round trip, ask the conflict to resolve itself with an update:

fn get_or_insert_tag(r: Repo, name: String) -> Result(Tag, LodeError) {
  use inserted <- result.try(repo.upsert(
    repo: r,
    schema: tag_schema(),
    changeset: new_tag(name),
    on_conflict: on_conflict.Update(
      target: on_conflict.Columns([field.field("name")]),
      action: on_conflict.Set([#(field.field("name"), VString(name))]),
    ),
  ))
  option.to_result(inserted, error.NoResults)
}

This is Ecto’s on_conflict: [set: [name: name]], conflict_target: :name. Because the conflicting row is now updated rather than skipped, Postgres’s RETURNING clause always produces the stored row — id included — and repo.upsert reads it back, so the None branch is unreachable in practice. As in Ecto, the price is a write (and any triggers it fires) even when nothing actually changed.

A few details worth knowing:

Upserts and insert_all

We are still issuing one upsert per tag. Ecto’s guide finishes by batching them with insert_all plus :on_conflict; lode spells that repo.upsert_all — a bulk insert of plain typed rows (no changesets, no validations) with an upsert policy. Two queries now handle any number of tags: one batched upsert that skips every existing tag, then one read of all the tags by name.

import lode/query
import lode/query/expr.{type FieldRef, FieldRef}

// A typed accessor for tags.name at binding 0 (codegen can emit these).
fn tag_name() -> FieldRef(Tag, String) {
  FieldRef(0, "name", primitive.string())
}

fn insert_and_get_all_tags(
  r: Repo,
  names: List(String),
) -> Result(List(Tag), LodeError) {
  case names {
    [] -> Ok([])
    _ -> {
      let rows = list.map(names, fn(name) { Tag(id: 0, name: name) })
      use _written_count <- result.try(repo.upsert_all(
        repo: r,
        schema: tag_schema(),
        rows: rows,
        on_conflict: on_conflict.Nothing(target: on_conflict.Columns([field.field("name")])),
      ))
      repo.all(
        r,
        tag_schema(),
        query.from("tags", "t")
          |> query.where(expr.in_list(tag_name(), names)),
      )
    }
  }
}

create_post then swaps its per-tag loop for one call:

pub fn create_post(
  r: Repo,
  title title: String,
  body body: String,
  tags tags: String,
) -> Result(Post, LodeError) {
  use tag_rows <- result.try(
    insert_and_get_all_tags(r, parse_tag_names(tags)),
  )
  let cs =
    changeset.change(Post(id: 0, title: "", body: "", tags: []), post_schema())
    |> changeset.put_change(field.field("title"), VString(title))
    |> changeset.put_change(field.field("body"), VString(body))
    |> changeset.put_assoc("tags", list.map(tag_rows, fn(tag) {
      changeset.change(tag, tag_schema())
    }))
  repo.insert(r, post_schema(), cs)
}

The placeholder id: 0 in the bulk rows is harmless: id is autogenerated, so it is omitted from the insert and the database assigns real keys. upsert_all returns the number of rows actually written — skipped conflicts are not counted — and its sibling upsert_all_returning loads the written rows back. Neither returns the rows a Nothing policy skipped, which is why the follow-up repo.all with expr.in_list is still needed to collect all the tags, new and pre-existing alike (Ecto’s version has the same shape).

Ecto’s rendition of this example also needs :placeholders to share one timestamp across the batch, because insert_all does not autogenerate timestamps. lode sidesteps the issue: there is no automatic timestamping at all, so timestamp columns are best given database defaults in the migration and marked autogenerated in the schema — then bulk inserts simply omit them.

Closing remarks

Nothing above runs in a transaction, and that is deliberate: getting or upserting a tag is idempotent, so a failed create_post can be retried and the tag operations converge on the same result. The trade-off is the same one Ecto’s guide accepts — if the post insert fails, freshly created tags stick around unattached to any post. If orphaned tags matter to you, wrap insert_and_get_all_tags and the post insert in repo.transaction, or compose them with lode/multi.

For the read side of associations and preload, see Associations; for the full list of differences from Elixir’s Ecto, see Divergences from Ecto.

✨ Search Document