diff --git a/crates/db/src/fields.rs b/crates/db/src/fields.rs
new file mode 100644
index 0000000..d07ba6e
--- /dev/null
+++ b/crates/db/src/fields.rs
@@ -0,0 +1,123 @@
+//! Registry of flexible field definitions.
+
+use domain::{
+    AuthorityKind, FieldDefinition, FieldDefinitionId, FieldType, LocalizedLabel,
+    NewFieldDefinition, VocabularyId,
+};
+use sqlx::Row;
+
+/// Labels aggregated per row as JSON, to read a definition and its labels in one query.
+const LABELS_JSON: &str = "COALESCE(json_agg(json_build_object('lang', fdl.lang, 'label', fdl.label) \
+     ORDER BY fdl.lang) FILTER (WHERE fdl.field_definition_id IS NOT NULL), '[]'::json)";
+
+const SELECT_COLUMNS: &str =
+    "fd.id, fd.key, fd.data_type, fd.vocabulary_id, fd.authority_kind, fd.required, fd.group_key";
+
+/// Create a field definition and its labels. Multiple statements — pass a
+/// transaction connection (`&mut *tx`) for atomicity.
+pub async fn create_field_definition(
+    conn: &mut sqlx::PgConnection,
+    new: &NewFieldDefinition,
+) -> Result<FieldDefinitionId, sqlx::Error> {
+    let id = FieldDefinitionId::new();
+    let (data_type, vocabulary_id, authority_kind) = new.field_type.to_parts();
+
+    sqlx::query(
+        "INSERT INTO field_definition \
+         (id, key, data_type, vocabulary_id, authority_kind, required, group_key) \
+         VALUES ($1, $2, $3, $4, $5, $6, $7)",
+    )
+    .bind(id.to_uuid())
+    .bind(&new.key)
+    .bind(data_type)
+    .bind(vocabulary_id.map(|v| v.to_uuid()))
+    .bind(authority_kind.map(|k| k.as_str()))
+    .bind(new.required)
+    .bind(new.group_key.as_deref())
+    .execute(&mut *conn)
+    .await?;
+
+    for label in &new.labels {
+        sqlx::query(
+            "INSERT INTO field_definition_label (field_definition_id, lang, label) \
+             VALUES ($1, $2, $3)",
+        )
+        .bind(id.to_uuid())
+        .bind(&label.lang)
+        .bind(&label.label)
+        .execute(&mut *conn)
+        .await?;
+    }
+
+    Ok(id)
+}
+
+/// Look up a field definition by its key (with labels).
+pub async fn field_definition_by_key<'e, E>(
+    executor: E,
+    key: &str,
+) -> Result<Option<FieldDefinition>, sqlx::Error>
+where
+    E: sqlx::PgExecutor<'e>,
+{
+    let sql = format!(
+        "SELECT {SELECT_COLUMNS}, {LABELS_JSON} AS labels \
+         FROM field_definition fd \
+         LEFT JOIN field_definition_label fdl ON fdl.field_definition_id = fd.id \
+         WHERE fd.key = $1 GROUP BY fd.id"
+    );
+
+    let row = sqlx::query(&sql).bind(key).fetch_optional(executor).await?;
+
+    row.map(map_field_definition).transpose()
+}
+
+/// List all field definitions (with labels), ordered by key.
+pub async fn list_field_definitions<'e, E>(executor: E) -> Result<Vec<FieldDefinition>, sqlx::Error>
+where
+    E: sqlx::PgExecutor<'e>,
+{
+    let sql = format!(
+        "SELECT {SELECT_COLUMNS}, {LABELS_JSON} AS labels \
+         FROM field_definition fd \
+         LEFT JOIN field_definition_label fdl ON fdl.field_definition_id = fd.id \
+         GROUP BY fd.id ORDER BY fd.key"
+    );
+
+    let rows = sqlx::query(&sql).fetch_all(executor).await?;
+
+    rows.into_iter().map(map_field_definition).collect()
+}
+
+fn map_field_definition(row: sqlx::postgres::PgRow) -> Result<FieldDefinition, sqlx::Error> {
+    let data_type: String = row.try_get("data_type")?;
+    let vocabulary_id: Option<uuid::Uuid> = row.try_get("vocabulary_id")?;
+    let authority_kind: Option<String> = row.try_get("authority_kind")?;
+
+    let authority_kind = authority_kind
+        .map(|k| {
+            AuthorityKind::from_db(&k)
+                .ok_or_else(|| sqlx::Error::Decode(format!("unknown authority kind: {k}").into()))
+        })
+        .transpose()?;
+
+    let field_type = FieldType::from_parts(
+        &data_type,
+        vocabulary_id.map(VocabularyId::from_uuid),
+        authority_kind,
+    )
+    .ok_or_else(|| {
+        sqlx::Error::Decode(format!("inconsistent field type stored: {data_type}").into())
+    })?;
+
+    let labels: sqlx::types::Json<Vec<LocalizedLabel>> = row.try_get("labels")?;
+
+    Ok(FieldDefinition {
+        id: FieldDefinitionId::from_uuid(row.try_get("id")?),
+        key: row.try_get("key")?,
+        field_type,
+        required: row.try_get("required")?,
+        group_key: row.try_get("group_key")?,
+        labels: labels.0,
+    })
+}
diff --git a/crates/db/src/lib.rs b/crates/db/src/lib.rs
index 4623d8e..7e1475a 100644
--- a/crates/db/src/lib.rs
+++ b/crates/db/src/lib.rs
@@ -3,6 +3,7 @@
 pub mod audit;
 pub mod authority;
 pub mod catalog;
+pub mod fields;
 pub mod vocab;
 
 use sqlx::postgres::{PgPool, PgPoolOptions};
diff --git a/crates/db/tests/fields.rs b/crates/db/tests/fields.rs
new file mode 100644
index 0000000..d8a787b
--- /dev/null
+++ b/crates/db/tests/fields.rs
@@ -0,0 +1,118 @@
+use db::{Db, fields, vocab};
+use domain::{AuthorityKind, FieldType, LocalizedLabel, NewFieldDefinition};
+use sqlx::PgPool;
+
+fn labels() -> Vec<LocalizedLabel> {
+    vec![
+        LocalizedLabel {
+            lang: "sv".into(),
+            label: "material".into(),
+        },
+        LocalizedLabel {
+            lang: "en".into(),
+            label: "material".into(),
+        },
+    ]
+}
+
+#[sqlx::test]
+async fn text_field_round_trips(pool: PgPool) {
+    let db = Db::from_pool(pool);
+    let mut tx = db.pool().begin().await.unwrap();
+    let id = fields::create_field_definition(
+        &mut tx,
+        &NewFieldDefinition {
+            key: "comments".into(),
+            field_type: FieldType::Text,
+            required: false,
+            group_key: Some("identification".into()),
+            labels: labels(),
+        },
+    )
+    .await
+    .unwrap();
+    tx.commit().await.unwrap();
+
+    let def = fields::field_definition_by_key(db.pool(), "comments")
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(def.id, id);
+    assert_eq!(def.field_type, FieldType::Text);
+    assert_eq!(def.group_key.as_deref(), Some("identification"));
+    assert_eq!(def.labels.len(), 2);
+    assert!(
+        fields::field_definition_by_key(db.pool(), "nope")
+            .await
+            .unwrap()
+            .is_none()
+    );
+}
+
+#[sqlx::test]
+async fn term_and_authority_fields_round_trip_their_binding(pool: PgPool) {
+    let db = Db::from_pool(pool);
+    let material = vocab::create_vocabulary(db.pool(), "material")
+        .await
+        .unwrap();
+
+    let mut tx = db.pool().begin().await.unwrap();
+    fields::create_field_definition(
+        &mut tx,
+        &NewFieldDefinition {
+            key: "material".into(),
+            field_type: FieldType::Term {
+                vocabulary_id: material.id,
+            },
+            required: true,
+            group_key: None,
+            labels: labels(),
+        },
+    )
+    .await
+    .unwrap();
+    fields::create_field_definition(
+        &mut tx,
+        &NewFieldDefinition {
+            key: "maker".into(),
+            field_type: FieldType::Authority {
+                kind: Some(AuthorityKind::Person),
+            },
+            required: false,
+            group_key: None,
+            labels: vec![LocalizedLabel {
+                lang: "en".into(),
+                label: "maker".into(),
+            }],
+        },
+    )
+    .await
+    .unwrap();
+    tx.commit().await.unwrap();
+
+    let material_def = fields::field_definition_by_key(db.pool(), "material")
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(
+        material_def.field_type,
+        FieldType::Term {
+            vocabulary_id: material.id
+        }
+    );
+    assert!(material_def.required);
+
+    let maker_def = fields::field_definition_by_key(db.pool(), "maker")
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(
+        maker_def.field_type,
+        FieldType::Authority {
+            kind: Some(AuthorityKind::Person)
+        }
+    );
+
+    let all = fields::list_field_definitions(db.pool()).await.unwrap();
+    assert_eq!(all.len(), 2);
+}