diff --git a/components/service/pocket/README.md b/components/service/pocket/README.md
index 0520eff98c6..86c2e0ec752 100644
--- a/components/service/pocket/README.md
+++ b/components/service/pocket/README.md
@@ -9,17 +9,25 @@ Currently this supports:
 
 ## Usage
 1. For Pocket recommended stories:
-    - Use `PocketStoriesService#startPeriodicStoriesRefresh` and `PocketStoriesService#stopPeriodicStoriesRefresh`
-      as high up in the client app as possible (preferably in the Application object or in a single Activity) to ensure the
-      background story refresh functionality works for the entirety of the app lifetime.
-    - Use `PocketStoriesService.getStories` to get the current list of Pocket recommended stories.
+   - Use `PocketStoriesService#startPeriodicStoriesRefresh` and `PocketStoriesService#stopPeriodicStoriesRefresh`
+     as high up in the client app as possible (preferably in the Application object or in a single Activity) to ensure the
+     background story refresh functionality works for the entirety of the app lifetime.
+   - Use `PocketStoriesService.getStories` to get the current list of Pocket recommended stories.
 
 2. For Pocket sponsored stories:
-    - Use `PocketStoriesService#startPeriodicSponsoredStoriesRefresh` and `PocketStoriesService#stopPeriodicSponsoredStoriesRefresh`
-      as high up in the client app as possible (preferably in the Application object or in a single Activity) to ensure the
-      background story refresh functionality works for the entirety of the app lifetime.
-    - Use `PocketStoriesService.getSponsoredStories` to get the current list of Pocket recommended stories.
-    - Use `PocketStoriesService.deleteProfile` to delete all server stored information about the device to which sponsored stories were previously downloaded. This may include data like network ip and application tokens.
+   - Use `PocketStoriesService#startPeriodicSponsoredStoriesRefresh` and `PocketStoriesService#stopPeriodicSponsoredStoriesRefresh`
+     as high up in the client app as possible (preferably in the Application object or in a single Activity) to ensure the
+     background story refresh functionality works for the entirety of the app lifetime.
+   - Use `PocketStoriesService.getSponsoredStories` to get the current list of Pocket recommended stories.
+   - Use `PocketStoriesService,recordStoriesImpressions` to try and persist that a list of sponsored stories were shown to the user. (Safe to call even if those stories are not persisted).
+   - Use `PocketStoriesService.deleteProfile` to delete all server stored information about the device to which sponsored stories were previously downloaded. This may include data like network ip and application tokens.
+
+   ##### Pacing and rotating:
+   A new `PocketSponsoredStoryCaps` is available in the response from `PocketStoriesService.getSponsoredStories` which allows checking `currentImpressions`, `lifetimeCount`, `flightCount`, `flightPeriod` based on which the client can decide which stories to show.
+   All this is based on clients calling `PocketStoriesService,recordStoriesImpressions` to record new impressions in between application restarts.
+
+
+
 
 ### Setting up the dependency
 
diff --git a/components/service/pocket/build.gradle b/components/service/pocket/build.gradle
index 4074c68167b..25e7dd000ee 100644
--- a/components/service/pocket/build.gradle
+++ b/components/service/pocket/build.gradle
@@ -12,6 +12,7 @@ android {
     defaultConfig {
         minSdkVersion config.minSdkVersion
         targetSdkVersion config.targetSdkVersion
+        testInstrumentationRunner "androidx.test.runner.AndroidJUnitRunner"
 
         kapt {
             arguments {
@@ -29,6 +30,7 @@ android {
 
     sourceSets {
         test.assets.srcDirs += files("$projectDir/schemas".toString())
+        androidTest.assets.srcDirs += files("$projectDir/schemas".toString())
     }
 }
 
@@ -58,6 +60,14 @@ dependencies {
 
     testImplementation project(':support-test')
     testImplementation project(':lib-fetch-httpurlconnection')
+
+    androidTestImplementation project(':support-android-test')
+
+    androidTestImplementation Dependencies.androidx_room_testing
+    androidTestImplementation Dependencies.androidx_arch_core_testing
+    androidTestImplementation Dependencies.androidx_test_core
+    androidTestImplementation Dependencies.androidx_test_runner
+    androidTestImplementation Dependencies.androidx_test_rules
 }
 
 apply from: '../../../publish.gradle'
diff --git a/components/service/pocket/schemas/mozilla.components.service.pocket.stories.db.PocketRecommendationsDatabase/2.json b/components/service/pocket/schemas/mozilla.components.service.pocket.stories.db.PocketRecommendationsDatabase/2.json
index 963deb138b3..1b55ceee77b 100644
--- a/components/service/pocket/schemas/mozilla.components.service.pocket.stories.db.PocketRecommendationsDatabase/2.json
+++ b/components/service/pocket/schemas/mozilla.components.service.pocket.stories.db.PocketRecommendationsDatabase/2.json
@@ -2,7 +2,7 @@
   "formatVersion": 1,
   "database": {
     "version": 2,
-    "identityHash": "1ea41b5cc0791d92dd8f0db8b387fe6c",
+    "identityHash": "966f55824415a21a73640bd2641772f2",
     "entities": [
       {
         "tableName": "stories",
@@ -62,8 +62,14 @@
       },
       {
         "tableName": "spocs",
-        "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`url` TEXT NOT NULL, `title` TEXT NOT NULL, `imageUrl` TEXT NOT NULL, `sponsor` TEXT NOT NULL, `clickShim` TEXT NOT NULL, `impressionShim` TEXT NOT NULL, PRIMARY KEY(`url`))",
+        "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`id` INTEGER NOT NULL, `url` TEXT NOT NULL, `title` TEXT NOT NULL, `imageUrl` TEXT NOT NULL, `sponsor` TEXT NOT NULL, `clickShim` TEXT NOT NULL, `impressionShim` TEXT NOT NULL, `priority` INTEGER NOT NULL, `lifetimeCapCount` INTEGER NOT NULL, `flightCapCount` INTEGER NOT NULL, `flightCapPeriod` INTEGER NOT NULL, PRIMARY KEY(`id`))",
         "fields": [
+          {
+            "fieldPath": "id",
+            "columnName": "id",
+            "affinity": "INTEGER",
+            "notNull": true
+          },
           {
             "fieldPath": "url",
             "columnName": "url",
@@ -99,22 +105,90 @@
             "columnName": "impressionShim",
             "affinity": "TEXT",
             "notNull": true
+          },
+          {
+            "fieldPath": "priority",
+            "columnName": "priority",
+            "affinity": "INTEGER",
+            "notNull": true
+          },
+          {
+            "fieldPath": "lifetimeCapCount",
+            "columnName": "lifetimeCapCount",
+            "affinity": "INTEGER",
+            "notNull": true
+          },
+          {
+            "fieldPath": "flightCapCount",
+            "columnName": "flightCapCount",
+            "affinity": "INTEGER",
+            "notNull": true
+          },
+          {
+            "fieldPath": "flightCapPeriod",
+            "columnName": "flightCapPeriod",
+            "affinity": "INTEGER",
+            "notNull": true
           }
         ],
         "primaryKey": {
           "columnNames": [
-            "url"
+            "id"
           ],
           "autoGenerate": false
         },
         "indices": [],
         "foreignKeys": []
+      },
+      {
+        "tableName": "spocs_impressions",
+        "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`spocId` INTEGER NOT NULL, `impressionId` INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL, `impressionDateInSeconds` INTEGER NOT NULL, FOREIGN KEY(`spocId`) REFERENCES `spocs`(`id`) ON UPDATE NO ACTION ON DELETE CASCADE )",
+        "fields": [
+          {
+            "fieldPath": "spocId",
+            "columnName": "spocId",
+            "affinity": "INTEGER",
+            "notNull": true
+          },
+          {
+            "fieldPath": "impressionId",
+            "columnName": "impressionId",
+            "affinity": "INTEGER",
+            "notNull": true
+          },
+          {
+            "fieldPath": "impressionDateInSeconds",
+            "columnName": "impressionDateInSeconds",
+            "affinity": "INTEGER",
+            "notNull": true
+          }
+        ],
+        "primaryKey": {
+          "columnNames": [
+            "impressionId"
+          ],
+          "autoGenerate": true
+        },
+        "indices": [],
+        "foreignKeys": [
+          {
+            "table": "spocs",
+            "onDelete": "CASCADE",
+            "onUpdate": "NO ACTION",
+            "columns": [
+              "spocId"
+            ],
+            "referencedColumns": [
+              "id"
+            ]
+          }
+        ]
       }
     ],
     "views": [],
     "setupQueries": [
       "CREATE TABLE IF NOT EXISTS room_master_table (id INTEGER PRIMARY KEY,identity_hash TEXT)",
-      "INSERT OR REPLACE INTO room_master_table (id,identity_hash) VALUES(42, '1ea41b5cc0791d92dd8f0db8b387fe6c')"
+      "INSERT OR REPLACE INTO room_master_table (id,identity_hash) VALUES(42, '966f55824415a21a73640bd2641772f2')"
     ]
   }
 }
\ No newline at end of file
diff --git a/components/service/pocket/src/androidTest/java/mozilla/components/service/pocket/stories/db/PocketRecommendationsDatabaseTest.kt b/components/service/pocket/src/androidTest/java/mozilla/components/service/pocket/stories/db/PocketRecommendationsDatabaseTest.kt
new file mode 100644
index 00000000000..f15ecaa0e79
--- /dev/null
+++ b/components/service/pocket/src/androidTest/java/mozilla/components/service/pocket/stories/db/PocketRecommendationsDatabaseTest.kt
@@ -0,0 +1,212 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
+
+package mozilla.components.service.pocket.stories.db
+
+import android.content.Context
+import androidx.arch.core.executor.testing.InstantTaskExecutorRule
+import androidx.room.Room
+import androidx.room.testing.MigrationTestHelper
+import androidx.sqlite.db.framework.FrameworkSQLiteOpenHelperFactory
+import androidx.test.core.app.ApplicationProvider
+import androidx.test.platform.app.InstrumentationRegistry
+import kotlinx.coroutines.runBlocking
+import mozilla.components.service.pocket.spocs.db.SpocEntity
+import mozilla.components.service.pocket.spocs.db.SpocImpressionEntity
+import mozilla.components.service.pocket.stories.db.PocketRecommendationsDatabase.Companion
+import org.junit.After
+import org.junit.Assert.assertEquals
+import org.junit.Before
+import org.junit.Rule
+import org.junit.Test
+import java.util.concurrent.ExecutorService
+import java.util.concurrent.Executors
+
+private const val MIGRATION_TEST_DB = "migration-test"
+
+class PocketRecommendationsDatabaseTest {
+    private lateinit var context: Context
+    private lateinit var executor: ExecutorService
+    private lateinit var database: PocketRecommendationsDatabase
+
+    @get:Rule
+    @Suppress("DEPRECATION")
+    val helper: MigrationTestHelper = MigrationTestHelper(
+        InstrumentationRegistry.getInstrumentation(),
+        PocketRecommendationsDatabase::class.java.canonicalName,
+        FrameworkSQLiteOpenHelperFactory()
+    )
+
+    @get:Rule
+    var instantTaskExecutorRule = InstantTaskExecutorRule()
+
+    @Before
+    fun setUp() {
+        executor = Executors.newSingleThreadExecutor()
+
+        context = ApplicationProvider.getApplicationContext()
+        database = Room.inMemoryDatabaseBuilder(context, PocketRecommendationsDatabase::class.java).build()
+    }
+
+    @After
+    fun tearDown() {
+        executor.shutdown()
+        database.close()
+    }
+
+    @Test
+    fun `test1To2MigrationAddsNewSpocsAndSpocsImpressionsTables`() = runBlocking {
+        // Create the database with the version 1 schema
+        val dbVersion1 = helper.createDatabase(MIGRATION_TEST_DB, 1).apply {
+            execSQL(
+                "INSERT INTO " +
+                    "'${Companion.TABLE_NAME_STORIES}' " +
+                    "(url, title, imageUrl, publisher, category, timeToRead, timesShown) " +
+                    "VALUES (" +
+                    "'${story.url}'," +
+                    "'${story.title}'," +
+                    "'${story.imageUrl}'," +
+                    "'${story.publisher}'," +
+                    "'${story.category}'," +
+                    "'${story.timeToRead}'," +
+                    "'${story.timesShown}'" +
+                    ")"
+            )
+        }
+        // Validate the persisted data which will be re-checked after migration
+        dbVersion1.query(
+            "SELECT * FROM ${Companion.TABLE_NAME_STORIES}"
+        ).use { cursor ->
+            assertEquals(1, cursor.count)
+
+            cursor.moveToFirst()
+            assertEquals(
+                story,
+                PocketStoryEntity(
+                    url = cursor.getString(0),
+                    title = cursor.getString(1),
+                    imageUrl = cursor.getString(2),
+                    publisher = cursor.getString(3),
+                    category = cursor.getString(4),
+                    timeToRead = cursor.getInt(5),
+                    timesShown = cursor.getLong(6),
+                )
+            )
+        }
+
+        val impression = SpocImpressionEntity(spoc.id).apply {
+            impressionId = 1
+            impressionDateInSeconds = 700L
+        }
+        // Migrate the initial database to the version 2 schema
+        val dbVersion2 = helper.runMigrationsAndValidate(
+            MIGRATION_TEST_DB, 2, true, Migrations.migration_1_2
+        ).apply {
+            execSQL(
+                "INSERT INTO " +
+                    "'${Companion.TABLE_NAME_SPOCS}' (" +
+                    "id, url, title, imageUrl, sponsor, clickShim, impressionShim, " +
+                    "priority, lifetimeCapCount, flightCapCount, flightCapPeriod" +
+                    ") VALUES (" +
+                    "'${spoc.id}'," +
+                    "'${spoc.url}'," +
+                    "'${spoc.title}'," +
+                    "'${spoc.imageUrl}'," +
+                    "'${spoc.sponsor}'," +
+                    "'${spoc.clickShim}'," +
+                    "'${spoc.impressionShim}'," +
+                    "'${spoc.priority}'," +
+                    "'${spoc.lifetimeCapCount}'," +
+                    "'${spoc.flightCapCount}'," +
+                    "'${spoc.flightCapPeriod}'" +
+                    ")"
+            )
+
+            execSQL(
+                "INSERT INTO " +
+                    "'${Companion.TABLE_NAME_SPOCS_IMPRESSIONS}' (" +
+                    "spocId, impressionId, impressionDateInSeconds" +
+                    ") VALUES (" +
+                    "'${impression.spocId}'," +
+                    "'${impression.impressionId}'," +
+                    "'${impression.impressionDateInSeconds}'" +
+                    ")"
+            )
+        }
+        // Re-check the initial data we had
+        dbVersion2.query(
+            "SELECT * FROM ${Companion.TABLE_NAME_STORIES}"
+        ).use { cursor ->
+            assertEquals(1, cursor.count)
+
+            cursor.moveToFirst()
+            assertEquals(
+                story,
+                PocketStoryEntity(
+                    url = cursor.getString(0),
+                    title = cursor.getString(1),
+                    imageUrl = cursor.getString(2),
+                    publisher = cursor.getString(3),
+                    category = cursor.getString(4),
+                    timeToRead = cursor.getInt(5),
+                    timesShown = cursor.getLong(6),
+                )
+            )
+        }
+        // Finally validate that the new spocs are persisted successfully
+        dbVersion2.query(
+            "SELECT * FROM ${Companion.TABLE_NAME_SPOCS}"
+        ).use { cursor ->
+            assertEquals(1, cursor.count)
+
+            cursor.moveToFirst()
+            assertEquals(spoc.id, cursor.getInt(0))
+            assertEquals(spoc.url, cursor.getString(1))
+            assertEquals(spoc.title, cursor.getString(2))
+            assertEquals(spoc.imageUrl, cursor.getString(3))
+            assertEquals(spoc.sponsor, cursor.getString(4))
+            assertEquals(spoc.clickShim, cursor.getString(5))
+            assertEquals(spoc.impressionShim, cursor.getString(6))
+            assertEquals(spoc.priority, cursor.getInt(7))
+            assertEquals(spoc.lifetimeCapCount, cursor.getInt(8))
+            assertEquals(spoc.flightCapCount, cursor.getInt(9))
+            assertEquals(spoc.flightCapPeriod, cursor.getInt(10))
+        }
+        // And that the impression was also persisted successfully
+        dbVersion2.query(
+            "SELECT * FROM ${Companion.TABLE_NAME_SPOCS_IMPRESSIONS}"
+        ).use { cursor ->
+            assertEquals(1, cursor.count)
+
+            cursor.moveToFirst()
+            assertEquals(impression.spocId, cursor.getInt(0))
+            assertEquals(impression.impressionId, cursor.getInt(1))
+            assertEquals(impression.impressionDateInSeconds, cursor.getLong(2))
+        }
+    }
+}
+
+private val story = PocketStoryEntity(
+    title = "How to Get Rid of Black Mold Naturally",
+    url = "https://getpocket.com/explore/item/how-to-get-rid-of-black-mold-naturally",
+    imageUrl = "https://img-getpocket.cdn.mozilla.net/{wh}/filters:format(jpeg):quality(60):no_upscale():strip_exif()/https%3A%2F%2Fpocket-image-cache.com%2F1200x%2Ffilters%3Aformat(jpg)%3Aextract_focal()%2Fhttps%253A%252F%252Fpocket-syndicated-images.s3.amazonaws.com%252Farticles%252F6757%252F1628024495_6109ae86db6cc.png",
+    publisher = "Pocket",
+    category = "general",
+    timeToRead = 4,
+    timesShown = 23
+)
+
+private val spoc = SpocEntity(
+    id = 191739319,
+    url = "https://i.geistm.com/l/GC_7ReasonsKetoV2_Journiest?bcid=601c567ac5b18a0414cce1d4&bhid=624f3ea9adad7604086ac6b3&utm_content=PKT_A_7ReasonsKetoV2_Journiest_40702022_RawMeatballUGC_130Off_601c567ac5b18a0414cce1d4_624f3ea9adad7604086ac6b3&tv=su4&ct=NAT-PK-PROS-130OFF5WEEK-037&utm_medium=DB&utm_source=pocket~geistm&utm_campaign=PKT_A_7ReasonsKetoV2_Journiest_40702022_RawMeatballUGC_130Off",
+    title = "Eating Keto Has Never Been So Easy With Green Chef",
+    imageUrl = "https://img-getpocket.cdn.mozilla.net/direct?url=realUrl.png&resize=w618-h310",
+    sponsor = "Green Chef",
+    clickShim = "193815086ClickShim",
+    impressionShim = "193815086ImpressionShim",
+    priority = 3,
+    lifetimeCapCount = 50,
+    flightCapCount = 10,
+    flightCapPeriod = 86400,
+)
diff --git a/components/service/pocket/src/main/java/mozilla/components/service/pocket/PocketStoriesService.kt b/components/service/pocket/src/main/java/mozilla/components/service/pocket/PocketStoriesService.kt
index 50d0377a988..292b05d92bd 100644
--- a/components/service/pocket/src/main/java/mozilla/components/service/pocket/PocketStoriesService.kt
+++ b/components/service/pocket/src/main/java/mozilla/components/service/pocket/PocketStoriesService.kt
@@ -151,4 +151,13 @@ class PocketStoriesService(
     suspend fun updateStoriesTimesShown(updatedStories: List<PocketRecommendedStory>) {
         storiesUseCases.updateTimesShown(updatedStories)
     }
+
+    /**
+     * Persist locally that the sponsored Pocket stories containing the ids from [storiesShown]
+     * were shown to the user.
+     * This is safe to call with any ids, even ones for stories not currently persisted anymore.
+     */
+    suspend fun recordStoriesImpressions(storiesShown: List<Int>) {
+        spocsUseCases?.recordImpression?.invoke(storiesShown)
+    }
 }
diff --git a/components/service/pocket/src/main/java/mozilla/components/service/pocket/PocketStory.kt b/components/service/pocket/src/main/java/mozilla/components/service/pocket/PocketStory.kt
index d8b03353a77..a14a9a9e6ca 100644
--- a/components/service/pocket/src/main/java/mozilla/components/service/pocket/PocketStory.kt
+++ b/components/service/pocket/src/main/java/mozilla/components/service/pocket/PocketStory.kt
@@ -42,6 +42,7 @@ sealed class PocketStory {
     /**
      * A Pocket sponsored story.
      *
+     * @property id Unique id of this story.
      * @property title The title of the story.
      * @property url 3rd party url containing the original story.
      * @property imageUrl A url to a still image representing the story.
@@ -49,13 +50,19 @@ sealed class PocketStory {
      * with a specific resolution and the CENTER_CROP ScaleType.
      * @property sponsor 3rd party sponsor of this story, e.g. "NextAdvisor".
      * @property shim Unique identifiers for when the user interacts with this story.
+     * @property priority Priority level in deciding which stories to be shown first.
+     * A lowest number means a higher priority.
+     * @property caps Story caps indented to control the maximum number of times the story should be shown.
      */
     data class PocketSponsoredStory(
+        val id: Int,
         override val title: String,
         override val url: String,
         val imageUrl: String,
         val sponsor: String,
         val shim: PocketSponsoredStoryShim,
+        val priority: Int,
+        val caps: PocketSponsoredStoryCaps,
     ) : PocketStory()
 
     /**
@@ -68,4 +75,25 @@ sealed class PocketStory {
         val click: String,
         val impression: String,
     )
+
+    /**
+     * Sponsored story caps indented to control the maximum number of times the story should be shown.
+     *
+     * @property currentImpressions List of all recorded impression of a sponsored Pocket story
+     * expressed in seconds from Epoch (as the result of `System.currentTimeMillis / 1000`).
+     * @property lifetimeCount Lifetime maximum number of times this story should be shown.
+     * This is independent from the count based on [flightCount] and [flightPeriod] and must never be reset.
+     * @property flightCount Maximum number of times this story should be shown in [flightPeriod].
+     * @property flightPeriod Period expressed as a number of seconds in which this story should be shown
+     * for at most [flightCount] times.
+     * Any time the period comes to an end the [flightCount] count should be restarted.
+     * Even if based on [flightCount] and [flightCount] this story can still be shown a couple more times
+     * if [lifetimeCount] was met then the story should not be shown anymore.
+     */
+    data class PocketSponsoredStoryCaps(
+        val currentImpressions: List<Long> = emptyList(),
+        val lifetimeCount: Int,
+        val flightCount: Int,
+        val flightPeriod: Int,
+    )
 }
diff --git a/components/service/pocket/src/main/java/mozilla/components/service/pocket/ext/Mappers.kt b/components/service/pocket/src/main/java/mozilla/components/service/pocket/ext/Mappers.kt
index b660f658353..f3c8527be78 100644
--- a/components/service/pocket/src/main/java/mozilla/components/service/pocket/ext/Mappers.kt
+++ b/components/service/pocket/src/main/java/mozilla/components/service/pocket/ext/Mappers.kt
@@ -7,6 +7,7 @@ package mozilla.components.service.pocket.ext
 import androidx.annotation.VisibleForTesting
 import mozilla.components.service.pocket.PocketStory.PocketRecommendedStory
 import mozilla.components.service.pocket.PocketStory.PocketSponsoredStory
+import mozilla.components.service.pocket.PocketStory.PocketSponsoredStoryCaps
 import mozilla.components.service.pocket.PocketStory.PocketSponsoredStoryShim
 import mozilla.components.service.pocket.spocs.api.ApiSpoc
 import mozilla.components.service.pocket.spocs.db.SpocEntity
@@ -53,25 +54,39 @@ internal fun PocketRecommendedStory.toPartialTimeShownUpdate(): PocketLocalStory
  */
 internal fun ApiSpoc.toLocalSpoc(): SpocEntity =
     SpocEntity(
+        id = flightId,
         url = url,
         title = title,
         imageUrl = imageSrc,
         sponsor = sponsor,
         clickShim = shim.click,
-        impressionShim = shim.impression
+        impressionShim = shim.impression,
+        priority = priority,
+        lifetimeCapCount = caps.lifetimeCount,
+        flightCapCount = caps.flightCount,
+        flightCapPeriod = caps.flightPeriod,
     )
 
 /**
  * Map Room entities to the object type that we expose to service clients.
  */
-internal fun SpocEntity.toPocketSponsoredStory(): PocketSponsoredStory =
-    PocketSponsoredStory(
-        title = title,
-        url = url,
-        imageUrl = imageUrl,
-        sponsor = sponsor,
-        shim = PocketSponsoredStoryShim(
-            click = clickShim,
-            impression = impressionShim
-        )
-    )
+internal fun SpocEntity.toPocketSponsoredStory(
+    impressions: List<Long> = emptyList()
+) = PocketSponsoredStory(
+    id = id,
+    title = title,
+    url = url,
+    imageUrl = imageUrl,
+    sponsor = sponsor,
+    shim = PocketSponsoredStoryShim(
+        click = clickShim,
+        impression = impressionShim
+    ),
+    priority = priority,
+    caps = PocketSponsoredStoryCaps(
+        currentImpressions = impressions,
+        lifetimeCount = lifetimeCapCount,
+        flightCount = flightCapCount,
+        flightPeriod = flightCapPeriod,
+    ),
+)
diff --git a/components/service/pocket/src/main/java/mozilla/components/service/pocket/ext/PocketStory.kt b/components/service/pocket/src/main/java/mozilla/components/service/pocket/ext/PocketStory.kt
new file mode 100644
index 00000000000..06695e74fcc
--- /dev/null
+++ b/components/service/pocket/src/main/java/mozilla/components/service/pocket/ext/PocketStory.kt
@@ -0,0 +1,50 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
+
+package mozilla.components.service.pocket.ext
+
+import mozilla.components.service.pocket.PocketStory.PocketSponsoredStory
+import mozilla.components.service.pocket.PocketStory.PocketSponsoredStoryCaps
+import java.util.concurrent.TimeUnit
+
+/**
+ * Get a list of all story impressions (expressed in seconds from Epoch) in the period between
+ * `now` down to [PocketSponsoredStoryCaps.flightPeriod].
+ */
+fun PocketSponsoredStory.getCurrentFlightImpressions(): List<Long> {
+    val now = TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis())
+    return caps.currentImpressions.filter {
+        now - it < caps.flightPeriod
+    }
+}
+
+/**
+ * Get if this story was already shown for the maximum number of times available in it's lifetime.
+ */
+fun PocketSponsoredStory.hasLifetimeImpressionsLimitReached(): Boolean {
+    return caps.currentImpressions.size >= caps.lifetimeCount
+}
+
+/**
+ * Get if this story was already shown for the maximum number of times available in the period
+ * specified by [PocketSponsoredStoryCaps.flightPeriod].
+ */
+fun PocketSponsoredStory.hasFlightImpressionsLimitReached(): Boolean {
+    return getCurrentFlightImpressions().size >= caps.flightCount
+}
+
+/**
+ * Record a new impression at this instant time and get this story back with updated impressions details.
+ * This only updates the in-memory data.
+ *
+ * It's recommended to use this method anytime a new impression needs to be recorded for a `PocketSponsoredStory`
+ * to ensure values consistency.
+ */
+fun PocketSponsoredStory.recordNewImpression(): PocketSponsoredStory {
+    return this.copy(
+        caps = caps.copy(
+            currentImpressions = caps.currentImpressions + TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis())
+        )
+    )
+}
diff --git a/components/service/pocket/src/main/java/mozilla/components/service/pocket/spocs/SpocsRepository.kt b/components/service/pocket/src/main/java/mozilla/components/service/pocket/spocs/SpocsRepository.kt
index 8f6a6d3ff71..a40764bdd77 100644
--- a/components/service/pocket/src/main/java/mozilla/components/service/pocket/spocs/SpocsRepository.kt
+++ b/components/service/pocket/src/main/java/mozilla/components/service/pocket/spocs/SpocsRepository.kt
@@ -10,6 +10,7 @@ import mozilla.components.service.pocket.PocketStory.PocketSponsoredStory
 import mozilla.components.service.pocket.ext.toLocalSpoc
 import mozilla.components.service.pocket.ext.toPocketSponsoredStory
 import mozilla.components.service.pocket.spocs.api.ApiSpoc
+import mozilla.components.service.pocket.spocs.db.SpocImpressionEntity
 import mozilla.components.service.pocket.stories.db.PocketRecommendationsDatabase
 
 /**
@@ -22,10 +23,18 @@ internal class SpocsRepository(context: Context) {
     internal val spocsDao by lazy { database.value.spocsDao() }
 
     /**
-     * Get the current locally persisted list of sponsored Pocket stories.
+     * Get the current locally persisted list of sponsored Pocket stories
+     * complete with the list of all locally persisted impressions data.
      */
     suspend fun getAllSpocs(): List<PocketSponsoredStory> {
-        return spocsDao.getAllSpocs().map { it.toPocketSponsoredStory() }
+        val spocs = spocsDao.getAllSpocs()
+        val impressions = spocsDao.getSpocsImpressions().groupBy { it.spocId }
+
+        return spocs.map {
+            it.toPocketSponsoredStory(
+                impressions[it.id]?.map { it.impressionDateInSeconds } ?: emptyList()
+            )
+        }
     }
 
     /**
@@ -43,4 +52,15 @@ internal class SpocsRepository(context: Context) {
     suspend fun addSpocs(spocs: List<ApiSpoc>) {
         spocsDao.cleanOldAndInsertNewSpocs(spocs.map { it.toLocalSpoc() })
     }
+
+    /**
+     * Add a new impression record for each of the spocs identified by the ids from [spocsShown].
+     * Will ignore adding new entries if the intended spocs are not persisted locally anymore.
+     * Recorded entries will automatically be cleaned when the spoc they target is deleted.
+     *
+     * @param spocsShown List of [PocketSponsoredStory.id] for which to record new impressions.
+     */
+    suspend fun recordImpressions(spocsShown: List<Int>) {
+        spocsDao.recordImpressions(spocsShown.map { SpocImpressionEntity(it) })
+    }
 }
diff --git a/components/service/pocket/src/main/java/mozilla/components/service/pocket/spocs/SpocsUseCases.kt b/components/service/pocket/src/main/java/mozilla/components/service/pocket/spocs/SpocsUseCases.kt
index bd870aa146a..0bc2b7485a1 100644
--- a/components/service/pocket/src/main/java/mozilla/components/service/pocket/spocs/SpocsUseCases.kt
+++ b/components/service/pocket/src/main/java/mozilla/components/service/pocket/spocs/SpocsUseCases.kt
@@ -7,6 +7,7 @@ package mozilla.components.service.pocket.spocs
 import android.content.Context
 import androidx.annotation.VisibleForTesting
 import mozilla.components.concept.fetch.Client
+import mozilla.components.service.pocket.PocketStory.PocketRecommendedStory
 import mozilla.components.service.pocket.PocketStory.PocketSponsoredStory
 import mozilla.components.service.pocket.spocs.api.SpocsEndpoint
 import mozilla.components.service.pocket.stories.api.PocketResponse.Failure
@@ -41,6 +42,10 @@ internal class SpocsUseCases(
         GetSponsoredStories(appContext)
     }
 
+    internal val recordImpression by lazy {
+        RecordImpression(appContext)
+    }
+
     /**
      * Delete all stored user data used for downloading sponsored stories.
      */
@@ -101,6 +106,25 @@ internal class SpocsUseCases(
         }
     }
 
+    /**
+     * Allows for atomically updating the [PocketRecommendedStory.timesShown] property of some recommended stories.
+     *
+     * @param context [Context] used for various system interactions and libraries initializations.
+     */
+    internal inner class RecordImpression(
+        @get:VisibleForTesting
+        internal val context: Context = this@SpocsUseCases.appContext
+    ) {
+        /**
+         * Update how many times certain stories were shown to the user.
+         */
+        suspend operator fun invoke(storiesShown: List<Int>) {
+            if (storiesShown.isNotEmpty()) {
+                getSpocsRepository(context).recordImpressions(storiesShown)
+            }
+        }
+    }
+
     /**
      * Allows deleting all stored user data used for downloading sponsored stories.
      *
diff --git a/components/service/pocket/src/main/java/mozilla/components/service/pocket/spocs/api/ApiSpoc.kt b/components/service/pocket/src/main/java/mozilla/components/service/pocket/spocs/api/ApiSpoc.kt
index e7186410976..bae665bc253 100644
--- a/components/service/pocket/src/main/java/mozilla/components/service/pocket/spocs/api/ApiSpoc.kt
+++ b/components/service/pocket/src/main/java/mozilla/components/service/pocket/spocs/api/ApiSpoc.kt
@@ -7,6 +7,7 @@ package mozilla.components.service.pocket.spocs.api
 /**
  * A Pocket sponsored as downloaded from the sponsored stories endpoint.
  *
+ * @property flightId Unique id of this story.
  * @property title the title of the story.
  * @property url 3rd party url containing the original story.
  * @property imageSrc a url to a still image representing the story.
@@ -14,13 +15,19 @@ package mozilla.components.service.pocket.spocs.api
  * with a specific resolution and the CENTER_CROP ScaleType.
  * @property sponsor 3rd party sponsor of this story, e.g. "NextAdvisor".
  * @property shim Unique identifiers for when the user interacts with this story.
+ * @property priority Priority level in deciding which stories to be shown first.
+ * A lowest number means a higher priority.
+ * @property caps Story caps indented to control the maximum number of times the story should be shown.
  */
 internal data class ApiSpoc(
+    val flightId: Int,
     val title: String,
     val url: String,
     val imageSrc: String,
     val sponsor: String,
     val shim: ApiSpocShim,
+    val priority: Int,
+    val caps: ApiSpocCaps,
 )
 
 /**
@@ -33,3 +40,21 @@ internal data class ApiSpocShim(
     val click: String,
     val impression: String,
 )
+
+/**
+ * Sponsored story caps indented to control the maximum number of times the story should be shown.
+ *
+ * @property lifetimeCount Lifetime maximum number of times this story should be shown.
+ * This is independent from the count based on [flightCount] and [flightPeriod] and must never be reset.
+ * @property flightCount Maximum number of times this story should be shown in [flightPeriod].
+ * @property flightPeriod Period expressed as a number of seconds in which this story should be shown
+ * for at most [flightCount] times.
+ * Any time the period comes to an end the [flightCount] count should be restarted.
+ * Even if based on [flightCount] and [flightCount] this story can still be shown a couple more times
+ * if [lifetimeCount] was met then the story should not be shown anymore.
+ */
+internal data class ApiSpocCaps(
+    val lifetimeCount: Int,
+    val flightCount: Int,
+    val flightPeriod: Int,
+)
diff --git a/components/service/pocket/src/main/java/mozilla/components/service/pocket/spocs/api/SpocsJSONParser.kt b/components/service/pocket/src/main/java/mozilla/components/service/pocket/spocs/api/SpocsJSONParser.kt
index 00eaa42f821..775e5037ddf 100644
--- a/components/service/pocket/src/main/java/mozilla/components/service/pocket/spocs/api/SpocsJSONParser.kt
+++ b/components/service/pocket/src/main/java/mozilla/components/service/pocket/spocs/api/SpocsJSONParser.kt
@@ -15,12 +15,24 @@ import org.json.JSONObject
 internal const val KEY_ARRAY_SPOCS = "spocs"
 @VisibleForTesting
 internal const val JSON_SPOC_SHIMS_KEY = "shim"
+@VisibleForTesting
+internal const val JSON_SPOC_CAPS_KEY = "caps"
+@VisibleForTesting
+internal const val JSON_SPOC_CAPS_LIFETIME_KEY = "lifetime"
+@VisibleForTesting
+internal const val JSON_SPOC_CAPS_FLIGHT_KEY = "campaign"
+@VisibleForTesting
+internal const val JSON_SPOC_CAPS_FLIGHT_COUNT_KEY = "count"
+@VisibleForTesting
+internal const val JSON_SPOC_CAPS_FLIGHT_PERIOD_KEY = "period"
+private const val JSON_SPOC_FLIGHT_ID_KEY = "flight_id"
 private const val JSON_SPOC_TITLE_KEY = "title"
 private const val JSON_SPOC_SPONSOR_KEY = "sponsor"
 private const val JSON_SPOC_URL_KEY = "url"
 private const val JSON_SPOC_IMAGE_SRC_KEY = "image_src"
 private const val JSON_SPOC_SHIM_CLICK_KEY = "click"
 private const val JSON_SPOC_SHIM_IMPRESSION_KEY = "impression"
+private const val JSON_SPOC_PRIORITY = "priority"
 
 /**
  * Holds functions that parse the JSON returned by the Pocket API and converts them to more usable Kotlin types.
@@ -43,11 +55,14 @@ internal object SpocsJSONParser {
 
     private fun jsonToSpoc(json: JSONObject): ApiSpoc? = try {
         ApiSpoc(
+            flightId = json.getInt(JSON_SPOC_FLIGHT_ID_KEY),
             title = json.getString(JSON_SPOC_TITLE_KEY),
             sponsor = json.getString(JSON_SPOC_SPONSOR_KEY),
             url = json.getString(JSON_SPOC_URL_KEY),
             imageSrc = json.getString(JSON_SPOC_IMAGE_SRC_KEY),
-            shim = jsonToShim(json.getJSONObject(JSON_SPOC_SHIMS_KEY))
+            shim = jsonToShim(json.getJSONObject(JSON_SPOC_SHIMS_KEY)),
+            priority = json.getInt(JSON_SPOC_PRIORITY),
+            caps = jsonToCaps(json.getJSONObject(JSON_SPOC_CAPS_KEY)),
         )
     } catch (e: JSONException) {
         null
@@ -57,4 +72,14 @@ internal object SpocsJSONParser {
         click = json.getString(JSON_SPOC_SHIM_CLICK_KEY),
         impression = json.getString(JSON_SPOC_SHIM_IMPRESSION_KEY)
     )
+
+    private fun jsonToCaps(json: JSONObject): ApiSpocCaps {
+        val flightCaps = json.getJSONObject(JSON_SPOC_CAPS_FLIGHT_KEY)
+
+        return ApiSpocCaps(
+            lifetimeCount = json.getInt(JSON_SPOC_CAPS_LIFETIME_KEY),
+            flightCount = flightCaps.getInt(JSON_SPOC_CAPS_FLIGHT_COUNT_KEY),
+            flightPeriod = flightCaps.getInt(JSON_SPOC_CAPS_FLIGHT_PERIOD_KEY)
+        )
+    }
 }
diff --git a/components/service/pocket/src/main/java/mozilla/components/service/pocket/spocs/db/SpocEntity.kt b/components/service/pocket/src/main/java/mozilla/components/service/pocket/spocs/db/SpocEntity.kt
index 14742b6da79..02c68b78458 100644
--- a/components/service/pocket/src/main/java/mozilla/components/service/pocket/spocs/db/SpocEntity.kt
+++ b/components/service/pocket/src/main/java/mozilla/components/service/pocket/spocs/db/SpocEntity.kt
@@ -10,14 +10,32 @@ import mozilla.components.service.pocket.stories.db.PocketRecommendationsDatabas
 
 /**
  * A sponsored Pocket story that is to be mapped to SQLite table.
+ *
+ * @property id Unique story id serving as the primary key of this entity.
+ * @property url URL where the original story can be read.
+ * @property title Title of the story.
+ * @property imageUrl URL of the hero image for this story.
+ * @property sponsor 3rd party sponsor of this story, e.g. "NextAdvisor".
+ * @property clickShim Telemetry identifier for when the sponsored story is clicked.
+ * @property impressionShim Telemetry identifier for when the sponsored story is seen by the user.
+ * @property priority Priority level in deciding which stories to be shown first.
+ * @property lifetimeCapCount Indicates how many times a sponsored story can be shown in total.
+ * @property flightCapCount Indicates how many times a sponsored story can be shown within a period.
+ * @property flightCapPeriod Indicates the period (number of seconds) in which at most [flightCapCount]
+ * stories can be shown.
  */
 @Entity(tableName = PocketRecommendationsDatabase.TABLE_NAME_SPOCS)
 internal data class SpocEntity(
     @PrimaryKey
+    val id: Int,
     val url: String,
     val title: String,
     val imageUrl: String,
     val sponsor: String,
     val clickShim: String,
     val impressionShim: String,
+    val priority: Int,
+    val lifetimeCapCount: Int,
+    val flightCapCount: Int,
+    val flightCapPeriod: Int,
 )
diff --git a/components/service/pocket/src/main/java/mozilla/components/service/pocket/spocs/db/SpocImpressionEntity.kt b/components/service/pocket/src/main/java/mozilla/components/service/pocket/spocs/db/SpocImpressionEntity.kt
new file mode 100644
index 00000000000..9d0f63a2a40
--- /dev/null
+++ b/components/service/pocket/src/main/java/mozilla/components/service/pocket/spocs/db/SpocImpressionEntity.kt
@@ -0,0 +1,41 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
+
+package mozilla.components.service.pocket.spocs.db
+
+import androidx.room.Entity
+import androidx.room.ForeignKey
+import androidx.room.PrimaryKey
+import mozilla.components.service.pocket.stories.db.PocketRecommendationsDatabase
+
+/**
+ * One sponsored Pocket story impression.
+ * Allows to easily create a relation between a particular spoc identified by it's [SpocEntity.id]
+ * and any number of impressions.
+ *
+ * @property spocId [SpocEntity.id] that this serves as an impression of.
+ * Used as a foreign key allowing to only add impressions for other persisted spocs and
+ * automatically remove all impressions when the spoc they refer to is deleted.
+ * @property impressionId Unique id of this entity. Primary key.
+ * @property impressionDateInSeconds Epoch based timestamp expressed in seconds (from System.currentTimeMillis / 1000)
+ * for when the spoc identified by [spocId] was shown to the user.
+ */
+@Entity(
+    tableName = PocketRecommendationsDatabase.TABLE_NAME_SPOCS_IMPRESSIONS,
+    foreignKeys = [
+        ForeignKey(
+            entity = SpocEntity::class,
+            parentColumns = arrayOf("id"),
+            childColumns = arrayOf("spocId"),
+            onDelete = ForeignKey.CASCADE,
+        )
+    ]
+)
+internal data class SpocImpressionEntity(
+    val spocId: Int,
+) {
+    @PrimaryKey(autoGenerate = true)
+    var impressionId: Int = 0
+    var impressionDateInSeconds: Long = System.currentTimeMillis() / 1000
+}
diff --git a/components/service/pocket/src/main/java/mozilla/components/service/pocket/spocs/db/SpocsDao.kt b/components/service/pocket/src/main/java/mozilla/components/service/pocket/spocs/db/SpocsDao.kt
index 00f1018972c..c74ddb00438 100644
--- a/components/service/pocket/src/main/java/mozilla/components/service/pocket/spocs/db/SpocsDao.kt
+++ b/components/service/pocket/src/main/java/mozilla/components/service/pocket/spocs/db/SpocsDao.kt
@@ -5,26 +5,69 @@
 package mozilla.components.service.pocket.spocs.db
 
 import androidx.room.Dao
+import androidx.room.Delete
 import androidx.room.Insert
 import androidx.room.OnConflictStrategy
 import androidx.room.Query
 import androidx.room.Transaction
 import mozilla.components.service.pocket.stories.db.PocketRecommendationsDatabase
+import java.util.concurrent.TimeUnit
 
 @Dao
 internal interface SpocsDao {
     @Transaction
     suspend fun cleanOldAndInsertNewSpocs(spocs: List<SpocEntity>) {
-        deleteAllSpocs()
+        val newSpocs = spocs.map { it.id }
+        val oldStoriesToDelete = getAllSpocs()
+            .filterNot { newSpocs.contains(it.id) }
+
+        deleteSpocs(oldStoriesToDelete)
         insertSpocs(spocs)
     }
 
     @Insert(onConflict = OnConflictStrategy.REPLACE) // Maybe some details changed
-    suspend fun insertSpocs(spocs: List<SpocEntity>)
+    suspend fun insertSpocs(stories: List<SpocEntity>)
+
+    @Transaction
+    suspend fun recordImpressions(stories: List<SpocImpressionEntity>) {
+        stories.forEach {
+            recordImpression(it.spocId, it.impressionDateInSeconds)
+        }
+    }
+
+    /**
+     * INSERT OR IGNORE method needed to prevent against "FOREIGN KEY constraint failed" exceptions
+     * if clients try to insert new impressions spocs not existing anymore in the database in cases where
+     * a different list of spocs were downloaded but the client operates with stale in-memory data.
+     *
+     * @param targetSpocId The `id` of the [SpocEntity] to add a new impression for.
+     * A new impression will be persisted only if a story with the indicated [targetSpocId] currently exists.
+     * @param targetImpressionDateInSeconds The timestamp expressed in seconds from Epoch for this impression.
+     * Defaults to the current time expressed in seconds as get from `System.currentTimeMillis / 1000`.
+     */
+    @Query(
+        "WITH newImpression(spocId, impressionDateInSeconds) AS (VALUES" +
+            "(:targetSpocId, :targetImpressionDateInSeconds)" +
+            ")" +
+            "INSERT INTO spocs_impressions(spocId, impressionDateInSeconds) " +
+            "SELECT impression.spocId, impression.impressionDateInSeconds " +
+            "FROM newImpression impression " +
+            "WHERE EXISTS (SELECT 1 FROM spocs spoc WHERE spoc.id = impression.spocId)"
+    )
+    suspend fun recordImpression(
+        targetSpocId: Int,
+        targetImpressionDateInSeconds: Long = TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis())
+    )
 
     @Query("DELETE FROM ${PocketRecommendationsDatabase.TABLE_NAME_SPOCS}")
     suspend fun deleteAllSpocs()
 
+    @Delete
+    suspend fun deleteSpocs(stories: List<SpocEntity>)
+
     @Query("SELECT * FROM ${PocketRecommendationsDatabase.TABLE_NAME_SPOCS}")
     suspend fun getAllSpocs(): List<SpocEntity>
+
+    @Query("SELECT * FROM ${PocketRecommendationsDatabase.TABLE_NAME_SPOCS_IMPRESSIONS}")
+    suspend fun getSpocsImpressions(): List<SpocImpressionEntity>
 }
diff --git a/components/service/pocket/src/main/java/mozilla/components/service/pocket/stories/db/PocketRecommendationsDatabase.kt b/components/service/pocket/src/main/java/mozilla/components/service/pocket/stories/db/PocketRecommendationsDatabase.kt
index de0808696ff..ab81eee5610 100644
--- a/components/service/pocket/src/main/java/mozilla/components/service/pocket/stories/db/PocketRecommendationsDatabase.kt
+++ b/components/service/pocket/src/main/java/mozilla/components/service/pocket/stories/db/PocketRecommendationsDatabase.kt
@@ -11,6 +11,7 @@ import androidx.room.RoomDatabase
 import androidx.room.migration.Migration
 import androidx.sqlite.db.SupportSQLiteDatabase
 import mozilla.components.service.pocket.spocs.db.SpocEntity
+import mozilla.components.service.pocket.spocs.db.SpocImpressionEntity
 import mozilla.components.service.pocket.spocs.db.SpocsDao
 
 /**
@@ -19,7 +20,8 @@ import mozilla.components.service.pocket.spocs.db.SpocsDao
 @Database(
     entities = [
         PocketStoryEntity::class,
-        SpocEntity::class
+        SpocEntity::class,
+        SpocImpressionEntity::class
     ],
     version = 2
 )
@@ -31,6 +33,7 @@ internal abstract class PocketRecommendationsDatabase : RoomDatabase() {
         private const val DATABASE_NAME = "pocket_recommendations"
         const val TABLE_NAME_STORIES = "stories"
         const val TABLE_NAME_SPOCS = "spocs"
+        const val TABLE_NAME_SPOCS_IMPRESSIONS = "spocs_impressions"
 
         @Volatile
         private var instance: PocketRecommendationsDatabase? = null
@@ -60,13 +63,30 @@ internal object Migrations {
             database.execSQL(
                 "CREATE TABLE IF NOT EXISTS " +
                     "`${PocketRecommendationsDatabase.TABLE_NAME_SPOCS}` (" +
+                    "`id` INTEGER NOT NULL, " +
                     "`url` TEXT NOT NULL, " +
                     "`title` TEXT NOT NULL, " +
                     "`imageUrl` TEXT NOT NULL, " +
                     "`sponsor` TEXT NOT NULL, " +
                     "`clickShim` TEXT NOT NULL, " +
                     "`impressionShim` TEXT NOT NULL, " +
-                    "PRIMARY KEY(`url`)" +
+                    "`priority` INTEGER NOT NULL, " +
+                    "`lifetimeCapCount` INTEGER NOT NULL, " +
+                    "`flightCapCount` INTEGER NOT NULL, " +
+                    "`flightCapPeriod` INTEGER NOT NULL, " +
+                    "PRIMARY KEY(`id`)" +
+                    ")"
+            )
+
+            database.execSQL(
+                "CREATE TABLE IF NOT EXISTS " +
+                    "`${PocketRecommendationsDatabase.TABLE_NAME_SPOCS_IMPRESSIONS}` (" +
+                    "`spocId` INTEGER NOT NULL, " +
+                    "`impressionId` INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL, " +
+                    "`impressionDateInSeconds` INTEGER NOT NULL, " +
+                    "FOREIGN KEY(`spocId`) " +
+                    "REFERENCES `spocs`(`id`) " +
+                    "ON UPDATE NO ACTION ON DELETE CASCADE " +
                     ")"
             )
         }
diff --git a/components/service/pocket/src/test/java/mozilla/components/service/pocket/PocketStoriesServiceTest.kt b/components/service/pocket/src/test/java/mozilla/components/service/pocket/PocketStoriesServiceTest.kt
index 55339bc0dc4..72f381f1be9 100644
--- a/components/service/pocket/src/test/java/mozilla/components/service/pocket/PocketStoriesServiceTest.kt
+++ b/components/service/pocket/src/test/java/mozilla/components/service/pocket/PocketStoriesServiceTest.kt
@@ -14,6 +14,7 @@ import mozilla.components.service.pocket.helpers.assertConstructorsVisibility
 import mozilla.components.service.pocket.spocs.SpocsUseCases
 import mozilla.components.service.pocket.spocs.SpocsUseCases.DeleteProfile
 import mozilla.components.service.pocket.spocs.SpocsUseCases.GetSponsoredStories
+import mozilla.components.service.pocket.spocs.SpocsUseCases.RecordImpression
 import mozilla.components.service.pocket.stories.PocketStoriesUseCases
 import mozilla.components.service.pocket.stories.PocketStoriesUseCases.GetPocketStories
 import mozilla.components.service.pocket.stories.PocketStoriesUseCases.UpdateStoriesTimesShown
@@ -175,4 +176,15 @@ class PocketStoriesServiceTest {
         val existingProfileResponse = service.deleteProfile()
         assertTrue(existingProfileResponse)
     }
+
+    @Test
+    fun `GIVEN PocketStoriesService WHEN recordStoriesImpressions THEN delegate to spocs useCases`() = runTest {
+        val recordImpressionsUseCase: RecordImpression = mock()
+        doReturn(recordImpressionsUseCase).`when`(spocsUseCases).recordImpression
+        val storiesIds = listOf(22, 33)
+
+        service.recordStoriesImpressions(storiesIds)
+
+        verify(recordImpressionsUseCase).invoke(storiesIds)
+    }
 }
diff --git a/components/service/pocket/src/test/java/mozilla/components/service/pocket/PocketStoryTest.kt b/components/service/pocket/src/test/java/mozilla/components/service/pocket/PocketStoryTest.kt
index 4641d1fa358..3e4eaae44e7 100644
--- a/components/service/pocket/src/test/java/mozilla/components/service/pocket/PocketStoryTest.kt
+++ b/components/service/pocket/src/test/java/mozilla/components/service/pocket/PocketStoryTest.kt
@@ -18,6 +18,11 @@ class PocketStoryTest {
         assertConstructorsVisibility(PocketSponsoredStory::class, KVisibility.PUBLIC)
     }
 
+    @Test
+    fun `GIVEN PocketSponsoredStoryCaps THEN it should be publicly available`() {
+        assertConstructorsVisibility(PocketRecommendedStory::class, KVisibility.PUBLIC)
+    }
+
     @Test
     fun `GIVEN PocketRecommendedStory THEN it should be publicly available`() {
         assertConstructorsVisibility(PocketRecommendedStory::class, KVisibility.PUBLIC)
@@ -48,7 +53,14 @@ class PocketStoryTest {
     @Test
     fun `GIVEN a PocketSponsoredStory WHEN it's title is accessed from parent THEN it returns the previously set value`() {
         val pocketRecommendedStory = PocketSponsoredStory(
-            title = "testTitle", url = "", imageUrl = "", sponsor = "", shim = mock()
+            id = 1,
+            title = "testTitle",
+            url = "",
+            imageUrl = "",
+            sponsor = "",
+            shim = mock(),
+            priority = 11,
+            caps = mock(),
         )
 
         val result = (pocketRecommendedStory as PocketStory).title
@@ -59,7 +71,14 @@ class PocketStoryTest {
     @Test
     fun `GIVEN a PocketSponsoredStory WHEN it's url is accessed from parent THEN it returns the previously set value`() {
         val pocketRecommendedStory = PocketSponsoredStory(
-            title = "", url = "testUrl", imageUrl = "", sponsor = "", shim = mock()
+            id = 2,
+            title = "",
+            url = "testUrl",
+            imageUrl = "",
+            sponsor = "",
+            shim = mock(),
+            priority = 33,
+            caps = mock(),
         )
 
         val result = (pocketRecommendedStory as PocketStory).url
diff --git a/components/service/pocket/src/test/java/mozilla/components/service/pocket/ext/MappersKtTest.kt b/components/service/pocket/src/test/java/mozilla/components/service/pocket/ext/MappersKtTest.kt
index 3174bee4684..0cf5c6a6bb4 100644
--- a/components/service/pocket/src/test/java/mozilla/components/service/pocket/ext/MappersKtTest.kt
+++ b/components/service/pocket/src/test/java/mozilla/components/service/pocket/ext/MappersKtTest.kt
@@ -8,6 +8,7 @@ import mozilla.components.service.pocket.helpers.PocketTestResources
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertNotEquals
 import org.junit.Assert.assertSame
+import org.junit.Assert.assertTrue
 import org.junit.Test
 import kotlin.reflect.full.memberProperties
 
@@ -78,12 +79,17 @@ class MappersKtTest {
 
         val result = apiStory.toLocalSpoc()
 
+        assertEquals(apiStory.flightId, result.id)
         assertSame(apiStory.title, result.title)
         assertSame(apiStory.url, result.url)
         assertSame(apiStory.imageSrc, result.imageUrl)
         assertSame(apiStory.sponsor, result.sponsor)
         assertSame(apiStory.shim.click, result.clickShim)
         assertSame(apiStory.shim.impression, result.impressionShim)
+        assertEquals(apiStory.priority, result.priority)
+        assertEquals(apiStory.caps.lifetimeCount, result.lifetimeCapCount)
+        assertEquals(apiStory.caps.flightCount, result.flightCapCount)
+        assertEquals(apiStory.caps.flightPeriod, result.flightCapPeriod)
     }
 
     @Test
@@ -92,11 +98,17 @@ class MappersKtTest {
 
         val result = localStory.toPocketSponsoredStory()
 
+        assertEquals(localStory.id, result.id)
         assertSame(localStory.title, result.title)
         assertSame(localStory.url, result.url)
         assertSame(localStory.imageUrl, result.imageUrl)
         assertSame(localStory.sponsor, result.sponsor)
         assertSame(localStory.clickShim, result.shim.click)
         assertSame(localStory.impressionShim, result.shim.impression)
+        assertEquals(localStory.priority, result.priority)
+        assertEquals(localStory.lifetimeCapCount, result.caps.lifetimeCount)
+        assertEquals(localStory.flightCapCount, result.caps.flightCount)
+        assertEquals(localStory.flightCapPeriod, result.caps.flightPeriod)
+        assertTrue(result.caps.currentImpressions.isEmpty())
     }
 }
diff --git a/components/service/pocket/src/test/java/mozilla/components/service/pocket/ext/PocketStoryKtTest.kt b/components/service/pocket/src/test/java/mozilla/components/service/pocket/ext/PocketStoryKtTest.kt
new file mode 100644
index 00000000000..f8bef315444
--- /dev/null
+++ b/components/service/pocket/src/test/java/mozilla/components/service/pocket/ext/PocketStoryKtTest.kt
@@ -0,0 +1,138 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
+
+package mozilla.components.service.pocket.ext
+
+import androidx.test.ext.junit.runners.AndroidJUnit4
+import mozilla.components.service.pocket.PocketStory.PocketSponsoredStory
+import mozilla.components.service.pocket.PocketStory.PocketSponsoredStoryCaps
+import mozilla.components.service.pocket.helpers.PocketTestResources
+import mozilla.components.support.test.mock
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertFalse
+import org.junit.Assert.assertSame
+import org.junit.Assert.assertTrue
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.mockito.Mockito.doReturn
+
+@RunWith(AndroidJUnit4::class)
+class PocketStoryKtTest {
+    private val nowInSeconds = System.currentTimeMillis() / 1000
+    private val flightPeriod = 100
+    private val flightImpression1 = nowInSeconds - flightPeriod / 2
+    private val flightImpression2 = nowInSeconds - flightPeriod / 3
+    private val currentImpressions = listOf(
+        nowInSeconds - flightPeriod * 2, // older impression that doesn't fit the flight period
+        flightImpression1,
+        flightImpression2
+    )
+
+    @Test
+    fun `GIVEN sponsored story impressions recorded WHEN asking for the current flight impression THEN return all impressions in flight period`() {
+        val storyCaps = PocketSponsoredStoryCaps(
+            currentImpressions = currentImpressions,
+            lifetimeCount = 10,
+            flightCount = 5,
+            flightPeriod = flightPeriod
+        )
+        val story: PocketSponsoredStory = mock()
+        doReturn(storyCaps).`when`(story).caps
+
+        val result = story.getCurrentFlightImpressions()
+
+        assertEquals(listOf(flightImpression1, flightImpression2), result)
+    }
+
+    @Test
+    fun `GIVEN sponsored story impressions recorded WHEN asking if lifetime impressions reached THEN return false if not`() {
+        val storyCaps = PocketSponsoredStoryCaps(
+            currentImpressions = currentImpressions,
+            lifetimeCount = 10,
+            flightCount = 5,
+            flightPeriod = flightPeriod
+        )
+        val story: PocketSponsoredStory = mock()
+        doReturn(storyCaps).`when`(story).caps
+
+        val result = story.hasLifetimeImpressionsLimitReached()
+
+        assertFalse(result)
+    }
+
+    @Test
+    fun `GIVEN sponsored story impressions recorded WHEN asking if lifetime impressions reached THEN return true if so`() {
+        val storyCaps = PocketSponsoredStoryCaps(
+            currentImpressions = currentImpressions,
+            lifetimeCount = 3,
+            flightCount = 3,
+            flightPeriod = flightPeriod
+        )
+        val story: PocketSponsoredStory = mock()
+        doReturn(storyCaps).`when`(story).caps
+
+        val result = story.hasLifetimeImpressionsLimitReached()
+
+        assertTrue(result)
+    }
+
+    @Test
+    fun `GIVEN sponsored story impressions recorded WHEN asking if flight impressions reached THEN return false if not`() {
+        val storyCaps = PocketSponsoredStoryCaps(
+            currentImpressions = currentImpressions,
+            lifetimeCount = 10,
+            flightCount = 5,
+            flightPeriod = flightPeriod
+        )
+        val story: PocketSponsoredStory = mock()
+        doReturn(storyCaps).`when`(story).caps
+
+        val result = story.hasFlightImpressionsLimitReached()
+
+        assertFalse(result)
+    }
+
+    @Test
+    fun `GIVEN sponsored story impressions recorded WHEN asking if flight impressions reached THEN return true if so`() {
+        val storyCaps = PocketSponsoredStoryCaps(
+            currentImpressions = currentImpressions,
+            lifetimeCount = 3,
+            flightCount = 2,
+            flightPeriod = flightPeriod
+        )
+        val story: PocketSponsoredStory = mock()
+        doReturn(storyCaps).`when`(story).caps
+
+        val result = story.hasFlightImpressionsLimitReached()
+
+        assertTrue(result)
+    }
+
+    @Test
+    fun `GIVEN a sponsored story WHEN recording a new impression THEN update the same story to contain a new impression recorded in seconds`() {
+        val story = PocketTestResources.dbExpectedPocketSpoc.toPocketSponsoredStory(currentImpressions)
+
+        assertEquals(3, story.caps.currentImpressions.size)
+        val result = story.recordNewImpression()
+
+        assertEquals(story.id, result.id)
+        assertSame(story.title, result.title)
+        assertSame(story.url, result.url)
+        assertSame(story.imageUrl, result.imageUrl)
+        assertSame(story.sponsor, result.sponsor)
+        assertSame(story.shim, result.shim)
+        assertEquals(story.priority, result.priority)
+        assertEquals(story.caps.lifetimeCount, result.caps.lifetimeCount)
+        assertEquals(story.caps.flightCount, result.caps.flightCount)
+        assertEquals(story.caps.flightPeriod, result.caps.flightPeriod)
+
+        assertEquals(4, result.caps.currentImpressions.size)
+        assertEquals(currentImpressions, result.caps.currentImpressions.take(3))
+        // Check if a new impression has been added for around this current time.
+        assertTrue(
+            LongRange(nowInSeconds - 5, nowInSeconds + 5)
+                .contains(result.caps.currentImpressions[3])
+        )
+    }
+}
diff --git a/components/service/pocket/src/test/java/mozilla/components/service/pocket/helpers/PocketTestResources.kt b/components/service/pocket/src/test/java/mozilla/components/service/pocket/helpers/PocketTestResources.kt
index 7bf53af776a..cb822a8d048 100644
--- a/components/service/pocket/src/test/java/mozilla/components/service/pocket/helpers/PocketTestResources.kt
+++ b/components/service/pocket/src/test/java/mozilla/components/service/pocket/helpers/PocketTestResources.kt
@@ -6,6 +6,7 @@ package mozilla.components.service.pocket.helpers
 
 import mozilla.components.service.pocket.PocketStory.PocketRecommendedStory
 import mozilla.components.service.pocket.spocs.api.ApiSpoc
+import mozilla.components.service.pocket.spocs.api.ApiSpocCaps
 import mozilla.components.service.pocket.spocs.api.ApiSpocShim
 import mozilla.components.service.pocket.spocs.db.SpocEntity
 import mozilla.components.service.pocket.stories.api.PocketApiStory
@@ -70,6 +71,7 @@ internal object PocketTestResources {
 
     val apiExpectedPocketSpocs: List<ApiSpoc> = listOf(
         ApiSpoc(
+            flightId = 191739319,
             title = "Eating Keto Has Never Been So Easy With Green Chef",
             url = "https://i.geistm.com/l/GC_7ReasonsKetoV2_Journiest?bcid=601c567ac5b18a0414cce1d4&bhid=624f3ea9adad7604086ac6b3&utm_content=PKT_A_7ReasonsKetoV2_Journiest_40702022_RawMeatballUGC_130Off_601c567ac5b18a0414cce1d4_624f3ea9adad7604086ac6b3&tv=su4&ct=NAT-PK-PROS-130OFF5WEEK-037&utm_medium=DB&utm_source=pocket~geistm&utm_campaign=PKT_A_7ReasonsKetoV2_Journiest_40702022_RawMeatballUGC_130Off",
             imageSrc = "https://img-getpocket.cdn.mozilla.net/direct?url=realUrl.png&resize=w618-h310",
@@ -77,9 +79,16 @@ internal object PocketTestResources {
             shim = ApiSpocShim(
                 click = "193815086ClickShim",
                 impression = "193815086ImpressionShim"
-            )
+            ),
+            priority = 3,
+            caps = ApiSpocCaps(
+                lifetimeCount = 50,
+                flightPeriod = 86400,
+                flightCount = 10,
+            ),
         ),
         ApiSpoc(
+            flightId = 191739667,
             title = "This Leading Cash Back Card Is a Slam Dunk if You Want a One-Card Wallet",
             url = "https://www.fool.com/the-ascent/credit-cards/landing/discover-it-cash-back-review-v2-csr/?utm_site=theascent&utm_campaign=ta-cc-co-pocket-discb-04012022-5-na-firefox&utm_medium=cpc&utm_source=pocket",
             imageSrc = "https://img-getpocket.cdn.mozilla.net/direct?url=https%3A//s.zkcdn.net/Advertisers/359f56a5423c4926ab3aa148e448d839.webp&resize=w618-h310",
@@ -87,9 +96,16 @@ internal object PocketTestResources {
             shim = ApiSpocShim(
                 click = "177986195ClickShim",
                 impression = "177986195ImpressionShim"
-            )
+            ),
+            priority = 2,
+            caps = ApiSpocCaps(
+                lifetimeCount = 50,
+                flightPeriod = 86400,
+                flightCount = 10,
+            ),
         ),
         ApiSpoc(
+            flightId = 189212196,
             title = "The Incredible Lawn Hack That Can Make Your Neighbors Green With Envy Over Your Lawn",
             url = "https://go.lawnbuddy.org/zf/50/7673?campaign=SUN_Pocket2022&creative=SUN_LawnCompare4-TheIncredibleLawnHackThatCanMakeYourNeighborsGreenWithEnvyOverYourLawn-WithoutSpendingAFortuneOnNewGrassAndWithoutBreakingASweat-20220420",
             imageSrc = "https://img-getpocket.cdn.mozilla.net/direct?url=https%3A//s.zkcdn.net/Advertisers/ce16302e184342cda0619c08b7604c9c.jpg&resize=w618-h310",
@@ -97,7 +113,13 @@ internal object PocketTestResources {
             shim = ApiSpocShim(
                 click = "192560056ClickShim",
                 impression = "192560056ImpressionShim"
-            )
+            ),
+            priority = 1,
+            caps = ApiSpocCaps(
+                lifetimeCount = 50,
+                flightPeriod = 86400,
+                flightCount = 10,
+            ),
         )
     )
 
@@ -122,11 +144,16 @@ internal object PocketTestResources {
     )
 
     val dbExpectedPocketSpoc = SpocEntity(
+        id = 191739319,
         url = "https://i.geistm.com/l/GC_7ReasonsKetoV2_Journiest?bcid=601c567ac5b18a0414cce1d4&bhid=624f3ea9adad7604086ac6b3&utm_content=PKT_A_7ReasonsKetoV2_Journiest_40702022_RawMeatballUGC_130Off_601c567ac5b18a0414cce1d4_624f3ea9adad7604086ac6b3&tv=su4&ct=NAT-PK-PROS-130OFF5WEEK-037&utm_medium=DB&utm_source=pocket~geistm&utm_campaign=PKT_A_7ReasonsKetoV2_Journiest_40702022_RawMeatballUGC_130Off",
         title = "Eating Keto Has Never Been So Easy With Green Chef",
         imageUrl = "https://img-getpocket.cdn.mozilla.net/direct?url=realUrl.png&resize=w618-h310",
         sponsor = "Green Chef",
         clickShim = "193815086ClickShim",
-        impressionShim = "193815086ImpressionShim"
+        impressionShim = "193815086ImpressionShim",
+        priority = 3,
+        lifetimeCapCount = 50,
+        flightCapCount = 10,
+        flightCapPeriod = 86400,
     )
 }
diff --git a/components/service/pocket/src/test/java/mozilla/components/service/pocket/spocs/SpocsRepositoryTest.kt b/components/service/pocket/src/test/java/mozilla/components/service/pocket/spocs/SpocsRepositoryTest.kt
index bc1a3f65167..8e3dda6f02c 100644
--- a/components/service/pocket/src/test/java/mozilla/components/service/pocket/spocs/SpocsRepositoryTest.kt
+++ b/components/service/pocket/src/test/java/mozilla/components/service/pocket/spocs/SpocsRepositoryTest.kt
@@ -8,15 +8,16 @@ import androidx.test.ext.junit.runners.AndroidJUnit4
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.test.runTest
 import mozilla.components.service.pocket.ext.toLocalSpoc
-import mozilla.components.service.pocket.ext.toPocketSponsoredStory
 import mozilla.components.service.pocket.helpers.PocketTestResources
+import mozilla.components.service.pocket.spocs.db.SpocImpressionEntity
 import mozilla.components.service.pocket.spocs.db.SpocsDao
+import mozilla.components.support.test.argumentCaptor
 import mozilla.components.support.test.robolectric.testContext
 import org.junit.Assert.assertEquals
+import org.junit.Assert.assertSame
 import org.junit.Before
 import org.junit.Test
 import org.junit.runner.RunWith
-import org.mockito.Mockito
 import org.mockito.Mockito.doReturn
 import org.mockito.Mockito.mock
 import org.mockito.Mockito.spy
@@ -31,19 +32,33 @@ class SpocsRepositoryTest {
     @Before
     fun setUp() {
         doReturn(dao).`when`(spocsRepo).spocsDao
-        Mockito.`when`(spocsRepo.spocsDao).thenReturn(dao)
     }
 
     @Test
     fun `GIVEN SpocsRepository WHEN asking for all spocs THEN return db entities mapped to domain type`() = runTest {
         val spoc = PocketTestResources.dbExpectedPocketSpoc
+        val impressions = listOf(
+            SpocImpressionEntity(spoc.id),
+            SpocImpressionEntity(333),
+            SpocImpressionEntity(spoc.id),
+        )
         doReturn(listOf(spoc)).`when`(dao).getAllSpocs()
+        doReturn(impressions).`when`(dao).getSpocsImpressions()
 
         val result = spocsRepo.getAllSpocs()
 
         verify(dao).getAllSpocs()
         assertEquals(1, result.size)
-        assertEquals(spoc.toPocketSponsoredStory(), result[0])
+        assertSame(spoc.title, result[0].title)
+        assertSame(spoc.url, result[0].url)
+        assertSame(spoc.imageUrl, result[0].imageUrl)
+        assertSame(spoc.impressionShim, result[0].shim.impression)
+        assertSame(spoc.clickShim, result[0].shim.click)
+        assertEquals(spoc.priority, result[0].priority)
+        assertEquals(2, result[0].caps.currentImpressions.size)
+        assertEquals(spoc.lifetimeCapCount, result[0].caps.lifetimeCount)
+        assertEquals(spoc.flightCapCount, result[0].caps.flightCount)
+        assertEquals(spoc.flightCapPeriod, result[0].caps.flightPeriod)
     }
 
     @Test
@@ -61,4 +76,18 @@ class SpocsRepositoryTest {
 
         verify(dao).cleanOldAndInsertNewSpocs(listOf(spoc.toLocalSpoc()))
     }
+
+    @Test
+    fun `GIVEN SpocsRepository WHEN recording new spocs impressions THEN add this to the database`() = runTest {
+        val spocsIds = listOf(3, 33, 444)
+        val impressionsCaptor = argumentCaptor<List<SpocImpressionEntity>>()
+
+        spocsRepo.recordImpressions(spocsIds)
+
+        verify(dao).recordImpressions(impressionsCaptor.capture())
+        assertEquals(spocsIds.size, impressionsCaptor.value.size)
+        assertEquals(spocsIds[0], impressionsCaptor.value[0].spocId)
+        assertEquals(spocsIds[1], impressionsCaptor.value[1].spocId)
+        assertEquals(spocsIds[2], impressionsCaptor.value[2].spocId)
+    }
 }
diff --git a/components/service/pocket/src/test/java/mozilla/components/service/pocket/spocs/SpocsUseCasesTest.kt b/components/service/pocket/src/test/java/mozilla/components/service/pocket/spocs/SpocsUseCasesTest.kt
index 64cfd6276ec..70f646b566b 100644
--- a/components/service/pocket/src/test/java/mozilla/components/service/pocket/spocs/SpocsUseCasesTest.kt
+++ b/components/service/pocket/src/test/java/mozilla/components/service/pocket/spocs/SpocsUseCasesTest.kt
@@ -18,6 +18,7 @@ import mozilla.components.service.pocket.stories.api.PocketResponse
 import mozilla.components.service.pocket.stories.api.PocketResponse.Failure
 import mozilla.components.service.pocket.stories.api.PocketResponse.Success
 import mozilla.components.support.test.any
+import mozilla.components.support.test.argumentCaptor
 import mozilla.components.support.test.mock
 import mozilla.components.support.test.robolectric.testContext
 import org.junit.Assert.assertEquals
@@ -177,6 +178,44 @@ class SpocsUseCasesTest {
         assertTrue(result.isEmpty())
     }
 
+    @Test
+    fun `GIVEN SpocsUseCases WHEN RecordImpression is constructed THEN use the same parameters`() {
+        val recordImpressionsUseCase = useCases.getStories
+
+        assertSame(testContext, recordImpressionsUseCase.context)
+    }
+
+    @Test
+    fun `GIVEN SpocsUseCases constructed WHEN RecordImpression is constructed separately THEN default to use the same parameters`() {
+        val recordImpressionsUseCase = useCases.RecordImpression()
+
+        assertSame(testContext, recordImpressionsUseCase.context)
+    }
+
+    @Test
+    fun `GIVEN SpocsUseCases constructed WHEN RecordImpression is constructed separately THEN allow using different parameters`() {
+        val context2: Context = mock()
+
+        val recordImpressionsUseCase = useCases.RecordImpression(context2)
+
+        assertSame(context2, recordImpressionsUseCase.context)
+    }
+
+    @Test
+    fun `GIVEN SpocsUseCases WHEN RecordImpression is called THEN record impressions in database`() = runTest {
+        val recordImpressionsUseCase = useCases.RecordImpression()
+        val storiesIds = listOf(5, 55, 4321)
+        val spocsIdsCaptor = argumentCaptor<List<Int>>()
+
+        recordImpressionsUseCase(storiesIds)
+
+        verify(spocsRepo).recordImpressions(spocsIdsCaptor.capture())
+        assertEquals(3, spocsIdsCaptor.value.size)
+        assertEquals(storiesIds[0], spocsIdsCaptor.value[0])
+        assertEquals(storiesIds[1], spocsIdsCaptor.value[1])
+        assertEquals(storiesIds[2], spocsIdsCaptor.value[2])
+    }
+
     @Test
     fun `GIVEN SpocsUseCases WHEN DeleteProfile is constructed THEN use the same parameters`() {
         val deleteProfileUseCase = useCases.deleteProfile
diff --git a/components/service/pocket/src/test/java/mozilla/components/service/pocket/spocs/api/SpocsJSONParserTest.kt b/components/service/pocket/src/test/java/mozilla/components/service/pocket/spocs/api/SpocsJSONParserTest.kt
index 0dd972cb788..a49d9bd96e3 100644
--- a/components/service/pocket/src/test/java/mozilla/components/service/pocket/spocs/api/SpocsJSONParserTest.kt
+++ b/components/service/pocket/src/test/java/mozilla/components/service/pocket/spocs/api/SpocsJSONParserTest.kt
@@ -111,6 +111,58 @@ class SpocsJSONParserTest {
         assertEquals(expectedSpocsIfMissingTitle.joinToString(), result.joinToString())
     }
 
+    @Test
+    fun `WHEN parsing spocs with missing priority THEN those entries are dropped`() {
+        val pocketJSON = PocketTestResources.pocketEndpointThreeSpocsResponse
+        val expectedSpocsIfMissingPriority = ArrayList(PocketTestResources.apiExpectedPocketSpocs)
+            .apply { removeAt(1) }
+        val pocketJsonWithMissingPriority = removeJsonFieldFromArrayIndex("priority", 1, pocketJSON)
+
+        val result = SpocsJSONParser.jsonToSpocs(pocketJsonWithMissingPriority)
+
+        assertEquals(2, result!!.size)
+        assertEquals(expectedSpocsIfMissingPriority.joinToString(), result.joinToString())
+    }
+
+    @Test
+    fun `WHEN parsing spocs with missing a lifetime count cap THEN those entries are dropped`() {
+        val pocketJSON = PocketTestResources.pocketEndpointThreeSpocsResponse
+        val expectedSpocsIfMissingLifetimeCap = ArrayList(PocketTestResources.apiExpectedPocketSpocs)
+            .apply { removeAt(0) }
+        val pocketJsonWithMissingLifetimeCap = removeCapFromSpoc(JSON_SPOC_CAPS_LIFETIME_KEY, 0, pocketJSON)
+
+        val result = SpocsJSONParser.jsonToSpocs(pocketJsonWithMissingLifetimeCap)
+
+        assertEquals(2, result!!.size)
+        assertEquals(expectedSpocsIfMissingLifetimeCap.joinToString(), result.joinToString())
+    }
+
+    @Test
+    fun `WHEN parsing spocs with missing a flight count cap THEN those entries are dropped`() {
+        val pocketJSON = PocketTestResources.pocketEndpointThreeSpocsResponse
+        val expectedSpocsIfMissingFlightCountCap = ArrayList(PocketTestResources.apiExpectedPocketSpocs)
+            .apply { removeAt(1) }
+        val pocketJsonWithMissingFlightCountCap = removeCapFromSpoc(JSON_SPOC_CAPS_FLIGHT_COUNT_KEY, 1, pocketJSON)
+
+        val result = SpocsJSONParser.jsonToSpocs(pocketJsonWithMissingFlightCountCap)
+
+        assertEquals(2, result!!.size)
+        assertEquals(expectedSpocsIfMissingFlightCountCap.joinToString(), result.joinToString())
+    }
+
+    @Test
+    fun `WHEN parsing spocs with missing a flight period cap THEN those entries are dropped`() {
+        val pocketJSON = PocketTestResources.pocketEndpointThreeSpocsResponse
+        val expectedSpocsIfMissingFlightPeriodCap = ArrayList(PocketTestResources.apiExpectedPocketSpocs)
+            .apply { removeAt(2) }
+        val pocketJsonWithMissingFlightPeriodCap = removeCapFromSpoc(JSON_SPOC_CAPS_FLIGHT_PERIOD_KEY, 2, pocketJSON)
+
+        val result = SpocsJSONParser.jsonToSpocs(pocketJsonWithMissingFlightPeriodCap)
+
+        assertEquals(2, result!!.size)
+        assertEquals(expectedSpocsIfMissingFlightPeriodCap.joinToString(), result.joinToString())
+    }
+
     @Test
     fun `WHEN parsing spocs for an invalid JSON String THEN null is returned`() {
         assertNull(SpocsJSONParser.jsonToSpocs("{!!}}"))
@@ -131,3 +183,18 @@ private fun removeShimFromSpoc(shimName: String, spocIndex: Int, json: String):
     spocJson.getJSONObject(JSON_SPOC_SHIMS_KEY).remove(shimName)
     return obj.toString()
 }
+
+private fun removeCapFromSpoc(cap: String, spocIndex: Int, json: String): String {
+    val obj = JSONObject(json)
+    val spocsJson = obj.getJSONArray(KEY_ARRAY_SPOCS)
+    val spocJson = spocsJson.getJSONObject(spocIndex)
+    val capsJSON = spocJson.getJSONObject(JSON_SPOC_CAPS_KEY)
+
+    if (cap == JSON_SPOC_CAPS_LIFETIME_KEY) {
+        capsJSON.remove(cap)
+    } else {
+        capsJSON.getJSONObject(JSON_SPOC_CAPS_FLIGHT_KEY).remove(cap)
+    }
+
+    return obj.toString()
+}
diff --git a/components/service/pocket/src/test/java/mozilla/components/service/pocket/spocs/db/SpocImpressionEntityTest.kt b/components/service/pocket/src/test/java/mozilla/components/service/pocket/spocs/db/SpocImpressionEntityTest.kt
new file mode 100644
index 00000000000..07be6ab8e42
--- /dev/null
+++ b/components/service/pocket/src/test/java/mozilla/components/service/pocket/spocs/db/SpocImpressionEntityTest.kt
@@ -0,0 +1,29 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
+
+package mozilla.components.service.pocket.spocs.db
+
+import mozilla.components.service.pocket.helpers.assertClassVisibility
+import org.junit.Assert.assertTrue
+import org.junit.Test
+import kotlin.reflect.KVisibility.INTERNAL
+
+class SpocImpressionEntityTest {
+    // This is the data type persisted locally. No need to be public
+    @Test
+    fun `GIVEN a spoc entity THEN it's visibility is internal`() {
+        assertClassVisibility(SpocImpressionEntity::class, INTERNAL)
+    }
+
+    @Test
+    fun `WHEN a new impression is created THEN the timestamp should be seconds from Epoch`() {
+        val nowInSeconds = System.currentTimeMillis() / 1000
+        val impression = SpocImpressionEntity(2)
+
+        assertTrue(
+            LongRange(nowInSeconds - 5, nowInSeconds + 5)
+                .contains(impression.impressionDateInSeconds)
+        )
+    }
+}
diff --git a/components/service/pocket/src/test/java/mozilla/components/service/pocket/spocs/db/SpocsDaoTest.kt b/components/service/pocket/src/test/java/mozilla/components/service/pocket/spocs/db/SpocsDaoTest.kt
index 6df23016eef..9f4e34d95cc 100644
--- a/components/service/pocket/src/test/java/mozilla/components/service/pocket/spocs/db/SpocsDaoTest.kt
+++ b/components/service/pocket/src/test/java/mozilla/components/service/pocket/spocs/db/SpocsDaoTest.kt
@@ -59,17 +59,31 @@ class SpocsDaoTest {
     }
 
     @Test
-    fun `GIVEN a story already persisted WHEN another story with different url is tried to be inserted THEN add that to the table`() = runTest {
+    fun `GIVEN a story already persisted WHEN another story with different id is tried to be inserted THEN add that to the table`() = runTest {
         val story = PocketTestResources.dbExpectedPocketSpoc
         val newStory = story.copy(
-            url = "updated" + story.url
+            id = 1
         )
         dao.insertSpocs(listOf(story))
 
         dao.insertSpocs(listOf(newStory))
         val result = dao.getAllSpocs()
 
-        assertEquals(listOf(story, newStory), result)
+        assertEquals(listOf(newStory, story), result)
+    }
+
+    @Test
+    fun `GIVEN a story already persisted WHEN another story with different url is tried to be inserted THEN replace the existing`() = runTest {
+        val story = PocketTestResources.dbExpectedPocketSpoc
+        val newStory = story.copy(
+            title = "updated" + story.url
+        )
+        dao.insertSpocs(listOf(story))
+
+        dao.insertSpocs(listOf(newStory))
+        val result = dao.getAllSpocs()
+
+        assertEquals(listOf(newStory), result)
     }
 
     @Test
@@ -142,12 +156,68 @@ class SpocsDaoTest {
         assertEquals(listOf(newStory), result)
     }
 
+    @Test
+    fun `GIVEN a story already persisted WHEN another story with different priority is tried to be inserted THEN replace the existing`() = runTest {
+        val story = PocketTestResources.dbExpectedPocketSpoc
+        val newStory = story.copy(
+            priority = 765
+        )
+        dao.insertSpocs(listOf(story))
+
+        dao.insertSpocs(listOf(newStory))
+        val result = dao.getAllSpocs()
+
+        assertEquals(listOf(newStory), result)
+    }
+
+    @Test
+    fun `GIVEN a story already persisted WHEN another story with a different lifetime cap count is tried to be inserted THEN replace the existing`() = runTest {
+        val story = PocketTestResources.dbExpectedPocketSpoc
+        val newStory = story.copy(
+            lifetimeCapCount = 123
+        )
+        dao.insertSpocs(listOf(story))
+
+        dao.insertSpocs(listOf(newStory))
+        val result = dao.getAllSpocs()
+
+        assertEquals(listOf(newStory), result)
+    }
+
+    @Test
+    fun `GIVEN a story already persisted WHEN another story with a different flight count cap is tried to be inserted THEN replace the existing`() = runTest {
+        val story = PocketTestResources.dbExpectedPocketSpoc
+        val newStory = story.copy(
+            flightCapCount = 999
+        )
+        dao.insertSpocs(listOf(story))
+
+        dao.insertSpocs(listOf(newStory))
+        val result = dao.getAllSpocs()
+
+        assertEquals(listOf(newStory), result)
+    }
+
+    @Test
+    fun `GIVEN a story already persisted WHEN another story with a different flight period cap is tried to be inserted THEN replace the existing`() = runTest {
+        val story = PocketTestResources.dbExpectedPocketSpoc
+        val newStory = story.copy(
+            flightCapPeriod = 1
+        )
+        dao.insertSpocs(listOf(story))
+
+        dao.insertSpocs(listOf(newStory))
+        val result = dao.getAllSpocs()
+
+        assertEquals(listOf(newStory), result)
+    }
+
     @Test
     fun `GIVEN no persisted storied WHEN asked to insert a list of stories THEN add them all to the table`() = runTest {
         val story1 = PocketTestResources.dbExpectedPocketSpoc
-        val story2 = PocketTestResources.dbExpectedPocketSpoc.copy(url = story1.url + "2")
-        val story3 = PocketTestResources.dbExpectedPocketSpoc.copy(url = story1.url + "3")
-        val story4 = PocketTestResources.dbExpectedPocketSpoc.copy(url = story1.url + "4")
+        val story2 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 2)
+        val story3 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 3)
+        val story4 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 4)
 
         dao.insertSpocs(listOf(story1, story2, story3, story4))
         val result = dao.getAllSpocs()
@@ -158,9 +228,9 @@ class SpocsDaoTest {
     @Test
     fun `GIVEN stories already persisted WHEN asked to delete them THEN remove all from the table`() = runTest {
         val story1 = PocketTestResources.dbExpectedPocketSpoc
-        val story2 = PocketTestResources.dbExpectedPocketSpoc.copy(url = story1.url + "2")
-        val story3 = PocketTestResources.dbExpectedPocketSpoc.copy(url = story1.url + "3")
-        val story4 = PocketTestResources.dbExpectedPocketSpoc.copy(url = story1.url + "4")
+        val story2 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 2)
+        val story3 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 3)
+        val story4 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 4)
         dao.insertSpocs(listOf(story1, story2, story3, story4))
 
         dao.deleteAllSpocs()
@@ -170,18 +240,262 @@ class SpocsDaoTest {
     }
 
     @Test
-    fun `GIVEN stories already persisted WHEN asked to cleanup and insert a new list THEN only persist the new list`() = runTest {
+    fun `GIVEN stories already persisted WHEN asked to delete some THEN remove remove the ones already persisted`() = runTest {
+        val story1 = PocketTestResources.dbExpectedPocketSpoc
+        val story2 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 2)
+        val story3 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 3)
+        val story4 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 4)
+        val story5 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 5)
+        dao.insertSpocs(listOf(story1, story2, story3, story4))
+
+        dao.deleteSpocs(listOf(story2, story3, story5))
+        val result = dao.getAllSpocs()
+
+        assertEquals(listOf(story1, story4), result)
+    }
+
+    @Test
+    fun `GIVEN stories already persisted WHEN asked to clean and insert new ones THEN remove from table all stories not found in the new list`() = runTest {
+        setupDatabseForTransactions()
+        val story1 = PocketTestResources.dbExpectedPocketSpoc
+        val story2 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 2)
+        val story3 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 3)
+        val story4 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 4)
+        dao.insertSpocs(listOf(story1, story2, story3, story4))
+
+        dao.cleanOldAndInsertNewSpocs(listOf(story2, story4))
+        val result = dao.getAllSpocs()
+
+        assertEquals(listOf(story2, story4), result)
+    }
+
+    @Test
+    fun `GIVEN stories already persisted WHEN asked to clean and insert new ones THEN update stories with new ids`() = runTest {
+        setupDatabseForTransactions()
+        val story1 = PocketTestResources.dbExpectedPocketSpoc
+        val story2 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 2)
+        val updatedStory1 = story1.copy(
+            id = story1.id * 234
+        )
+        dao.insertSpocs(listOf(story1, story2))
+
+        dao.cleanOldAndInsertNewSpocs(listOf(updatedStory1, story2))
+        val result = dao.getAllSpocs()
+
+        // Order gets reversed because the original story is replaced and another one is added.
+        assertEquals(listOf(story2, updatedStory1), result)
+    }
+
+    @Test
+    fun `GIVEN stories already persisted WHEN asked to clean and insert new ones THEN don't update story if only url changed`() = runTest {
+        setupDatabseForTransactions()
+        val story1 = PocketTestResources.dbExpectedPocketSpoc
+        val story2 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 2)
+        val updatedStory1 = story1.copy(
+            url = "updated" + story1.url
+        )
+        dao.insertSpocs(listOf(story1, story2))
+
+        dao.cleanOldAndInsertNewSpocs(listOf(updatedStory1, story2))
+        val result = dao.getAllSpocs()
+
+        assertEquals(listOf(updatedStory1, story2), result)
+    }
+
+    @Test
+    fun `GIVEN stories already persisted WHEN asked to clean and insert new ones THEN don't update story if only title changed`() = runTest {
+        setupDatabseForTransactions()
+        val story1 = PocketTestResources.dbExpectedPocketSpoc
+        val story2 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 2)
+        val updatedStory1 = story1.copy(
+            title = "updated" + story1.title
+        )
+        dao.insertSpocs(listOf(story1, story2))
+
+        dao.cleanOldAndInsertNewSpocs(listOf(updatedStory1, story2))
+        val result = dao.getAllSpocs()
+
+        assertEquals(listOf(updatedStory1, story2), result)
+    }
+
+    @Test
+    fun `GIVEN stories already persisted WHEN asked to clean and insert new ones THEN don't update story if only image url changed`() = runTest {
+        setupDatabseForTransactions()
+        val story1 = PocketTestResources.dbExpectedPocketSpoc
+        val story2 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 2)
+        val updatedStory1 = story1.copy(
+            imageUrl = "updated" + story1.imageUrl
+        )
+        dao.insertSpocs(listOf(story1, story2))
+
+        dao.cleanOldAndInsertNewSpocs(listOf(updatedStory1, story2))
+        val result = dao.getAllSpocs()
+
+        assertEquals(listOf(updatedStory1, story2), result)
+    }
+
+    @Test
+    fun `GIVEN stories already persisted WHEN asked to clean and insert new ones THEN don't update story if only sponsor changed`() = runTest {
+        setupDatabseForTransactions()
+        val story1 = PocketTestResources.dbExpectedPocketSpoc
+        val story2 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 2)
+        val updatedStory1 = story1.copy(
+            sponsor = "updated" + story1.sponsor
+        )
+        dao.insertSpocs(listOf(story1, story2))
+
+        dao.cleanOldAndInsertNewSpocs(listOf(updatedStory1, story2))
+        val result = dao.getAllSpocs()
+
+        assertEquals(listOf(updatedStory1, story2), result)
+    }
+
+    @Test
+    fun `GIVEN stories already persisted WHEN asked to clean and insert new ones THEN don't update story if only the click shim changed`() = runTest {
         setupDatabseForTransactions()
         val story1 = PocketTestResources.dbExpectedPocketSpoc
-        val story2 = PocketTestResources.dbExpectedPocketSpoc.copy(url = story1.url + "2")
-        val newStory = PocketTestResources.dbExpectedPocketSpoc.copy(url = "updated" + story1.url)
-        val newStory2 = PocketTestResources.dbExpectedPocketSpoc.copy(url = "updated" + story1.url + "2")
+        val story2 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 2)
+        val updatedStory1 = story1.copy(
+            clickShim = "updated" + story1.clickShim
+        )
         dao.insertSpocs(listOf(story1, story2))
 
-        dao.cleanOldAndInsertNewSpocs(listOf(newStory, newStory2))
+        dao.cleanOldAndInsertNewSpocs(listOf(updatedStory1, story2))
         val result = dao.getAllSpocs()
 
-        assertEquals(listOf(newStory, newStory2), result)
+        assertEquals(listOf(updatedStory1, story2), result)
+    }
+
+    @Test
+    fun `GIVEN stories already persisted WHEN asked to clean and insert new ones THEN don't update story if only the impression shim changed`() = runTest {
+        setupDatabseForTransactions()
+        val story1 = PocketTestResources.dbExpectedPocketSpoc
+        val story2 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 2)
+        val updatedStory1 = story1.copy(
+            impressionShim = "updated" + story1.impressionShim
+        )
+        dao.insertSpocs(listOf(story1, story2))
+
+        dao.cleanOldAndInsertNewSpocs(listOf(updatedStory1, story2))
+        val result = dao.getAllSpocs()
+
+        assertEquals(listOf(updatedStory1, story2), result)
+    }
+
+    @Test
+    fun `GIVEN stories already persisted WHEN asked to clean and insert new ones THEN don't update story if only priority changed`() = runTest {
+        setupDatabseForTransactions()
+        val story1 = PocketTestResources.dbExpectedPocketSpoc
+        val story2 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 2)
+        val updatedStory1 = story1.copy(
+            priority = 678
+        )
+        dao.insertSpocs(listOf(story1, story2))
+
+        dao.cleanOldAndInsertNewSpocs(listOf(updatedStory1, story2))
+        val result = dao.getAllSpocs()
+
+        assertEquals(listOf(updatedStory1, story2), result)
+    }
+
+    @Test
+    fun `GIVEN stories already persisted WHEN asked to clean and insert new ones THEN don't update story if only the lifetime count cap changed`() = runTest {
+        setupDatabseForTransactions()
+        val story1 = PocketTestResources.dbExpectedPocketSpoc
+        val story2 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 2)
+        val updatedStory1 = story1.copy(
+            lifetimeCapCount = 4322
+        )
+        dao.insertSpocs(listOf(story1, story2))
+
+        dao.cleanOldAndInsertNewSpocs(listOf(updatedStory1, story2))
+        val result = dao.getAllSpocs()
+
+        assertEquals(listOf(updatedStory1, story2), result)
+    }
+
+    @Test
+    fun `GIVEN stories already persisted WHEN asked to clean and insert new ones THEN don't update story if only the flight count cap changed`() = runTest {
+        setupDatabseForTransactions()
+        val story1 = PocketTestResources.dbExpectedPocketSpoc
+        val story2 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 2)
+        val updatedStory1 = story1.copy(
+            flightCapCount = 111111
+        )
+        dao.insertSpocs(listOf(story1, story2))
+
+        dao.cleanOldAndInsertNewSpocs(listOf(updatedStory1, story2))
+        val result = dao.getAllSpocs()
+
+        assertEquals(listOf(updatedStory1, story2), result)
+    }
+
+    @Test
+    fun `GIVEN stories already persisted WHEN asked to clean and insert new ones THEN don't update story if only the flight period cap changed`() = runTest {
+        setupDatabseForTransactions()
+        val story1 = PocketTestResources.dbExpectedPocketSpoc
+        val story2 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 2)
+        val updatedStory1 = story1.copy(
+            flightCapPeriod = 7
+        )
+        dao.insertSpocs(listOf(story1, story2))
+
+        dao.cleanOldAndInsertNewSpocs(listOf(updatedStory1, story2))
+        val result = dao.getAllSpocs()
+
+        assertEquals(listOf(updatedStory1, story2), result)
+    }
+
+    @Test
+    fun `GIVEN no stories are persisted WHEN asked to record an impression THEN don't persist data and don't throw errors`() = runTest {
+        dao.recordImpression(6543321)
+
+        val result = dao.getSpocsImpressions()
+
+        assertTrue(result.isEmpty())
+    }
+
+    @Test
+    fun `GIVEN stories are persisted WHEN asked to record impressions for other stories also THEN persist impression only for existing stories`() = runTest {
+        setupDatabseForTransactions()
+        val story1 = PocketTestResources.dbExpectedPocketSpoc
+        val story2 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 2)
+        val story3 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 3)
+        dao.insertSpocs(listOf(story1, story3))
+
+        dao.recordImpressions(
+            listOf(
+                SpocImpressionEntity(story1.id),
+                SpocImpressionEntity(story2.id),
+                SpocImpressionEntity(story3.id)
+            )
+        )
+        val result = dao.getSpocsImpressions()
+
+        assertEquals(2, result.size)
+        assertEquals(story1.id, result[0].spocId)
+        assertEquals(story3.id, result[1].spocId)
+    }
+
+    @Test
+    fun `GIVEN stories are persisted WHEN asked to record impressions for existing stories THEN persist the impressions`() = runTest {
+        setupDatabseForTransactions()
+        val story1 = PocketTestResources.dbExpectedPocketSpoc
+        val story2 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 2)
+        val story3 = PocketTestResources.dbExpectedPocketSpoc.copy(id = story1.id * 3)
+        dao.insertSpocs(listOf(story1, story2, story3))
+
+        dao.recordImpressions(
+            listOf(
+                SpocImpressionEntity(story1.id),
+                SpocImpressionEntity(story3.id)
+            )
+        )
+        val result = dao.getSpocsImpressions()
+
+        assertEquals(2, result.size)
+        assertEquals(story1.id, result[0].spocId)
+        assertEquals(story3.id, result[1].spocId)
     }
 
     /**
diff --git a/components/service/pocket/src/test/java/mozilla/components/service/pocket/stories/db/PocketRecommendationsDatabaseTest.kt b/components/service/pocket/src/test/java/mozilla/components/service/pocket/stories/db/PocketRecommendationsDatabaseTest.kt
deleted file mode 100644
index 126a427d468..00000000000
--- a/components/service/pocket/src/test/java/mozilla/components/service/pocket/stories/db/PocketRecommendationsDatabaseTest.kt
+++ /dev/null
@@ -1,133 +0,0 @@
-/* This Source Code Form is subject to the terms of the Mozilla Public
- * License, v. 2.0. If a copy of the MPL was not distributed with this
- * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
-
-package mozilla.components.service.pocket.stories.db
-
-import androidx.room.testing.MigrationTestHelper
-import androidx.sqlite.db.framework.FrameworkSQLiteOpenHelperFactory
-import androidx.test.ext.junit.runners.AndroidJUnit4
-import androidx.test.platform.app.InstrumentationRegistry
-import kotlinx.coroutines.ExperimentalCoroutinesApi
-import kotlinx.coroutines.test.runTest
-import mozilla.components.service.pocket.helpers.PocketTestResources
-import mozilla.components.service.pocket.spocs.db.SpocEntity
-import org.junit.Assert.assertEquals
-import org.junit.Rule
-import org.junit.Test
-import org.junit.runner.RunWith
-
-private const val MIGRATION_TEST_DB = "migration-test"
-
-@RunWith(AndroidJUnit4::class)
-class PocketRecommendationsDatabaseTest {
-    @get:Rule
-    @Suppress("DEPRECATION")
-    val helper: MigrationTestHelper = MigrationTestHelper(
-        InstrumentationRegistry.getInstrumentation(),
-        PocketRecommendationsDatabase::class.java.canonicalName,
-        FrameworkSQLiteOpenHelperFactory()
-    )
-
-    @OptIn(ExperimentalCoroutinesApi::class)
-    @Test
-    fun `GIVEN database at version 1 WHEN needing to migrate to version 2 THEN add a new spocs table`() = runTest {
-        val story = PocketTestResources.dbExpectedPocketStory
-        // Create the database with the version 1 schema
-        val dbVersion1 = helper.createDatabase(MIGRATION_TEST_DB, 1).apply {
-            execSQL(
-                "INSERT INTO " +
-                    "'${PocketRecommendationsDatabase.TABLE_NAME_STORIES}' " +
-                    "(url, title, imageUrl, publisher, category, timeToRead, timesShown) " +
-                    "VALUES (" +
-                    "'${story.url}'," +
-                    "'${story.title}'," +
-                    "'${story.imageUrl}'," +
-                    "'${story.publisher}'," +
-                    "'${story.category}'," +
-                    "'${story.timeToRead}'," +
-                    "'${story.timesShown}'" +
-                    ")"
-            )
-        }
-        // Validate the persisted data which will be re-checked after migration
-        dbVersion1.query(
-            "SELECT * FROM ${PocketRecommendationsDatabase.TABLE_NAME_STORIES}"
-        ).use { cursor ->
-            assertEquals(1, cursor.count)
-
-            cursor.moveToFirst()
-            assertEquals(
-                story,
-                PocketStoryEntity(
-                    url = cursor.getString(0),
-                    title = cursor.getString(1),
-                    imageUrl = cursor.getString(2),
-                    publisher = cursor.getString(3),
-                    category = cursor.getString(4),
-                    timeToRead = cursor.getInt(5),
-                    timesShown = cursor.getLong(6),
-                )
-            )
-        }
-
-        val spoc = PocketTestResources.dbExpectedPocketSpoc
-        // Migrate the initial database to the version 2 schema
-        val dbVersion2 = helper.runMigrationsAndValidate(
-            MIGRATION_TEST_DB, 2, true, Migrations.migration_1_2
-        ).apply {
-            execSQL(
-                "INSERT INTO " +
-                    "'${PocketRecommendationsDatabase.TABLE_NAME_SPOCS}' " +
-                    "(url, title, imageUrl, sponsor, clickShim, impressionShim) " +
-                    "VALUES (" +
-                    "'${spoc.url}'," +
-                    "'${spoc.title}'," +
-                    "'${spoc.imageUrl}'," +
-                    "'${spoc.sponsor}'," +
-                    "'${spoc.clickShim}'," +
-                    "'${spoc.impressionShim}'" +
-                    ")"
-            )
-        }
-        // Re-check the initial data we had
-        dbVersion2.query(
-            "SELECT * FROM ${PocketRecommendationsDatabase.TABLE_NAME_STORIES}"
-        ).use { cursor ->
-            assertEquals(1, cursor.count)
-
-            cursor.moveToFirst()
-            assertEquals(
-                story,
-                PocketStoryEntity(
-                    url = cursor.getString(0),
-                    title = cursor.getString(1),
-                    imageUrl = cursor.getString(2),
-                    publisher = cursor.getString(3),
-                    category = cursor.getString(4),
-                    timeToRead = cursor.getInt(5),
-                    timesShown = cursor.getLong(6),
-                )
-            )
-        }
-        // Finally validate that the new spocs are persisted successfully
-        dbVersion2.query(
-            "SELECT * FROM ${PocketRecommendationsDatabase.TABLE_NAME_SPOCS}"
-        ).use { cursor ->
-            assertEquals(1, cursor.count)
-
-            cursor.moveToFirst()
-            assertEquals(
-                spoc,
-                SpocEntity(
-                    url = cursor.getString(0),
-                    title = cursor.getString(1),
-                    imageUrl = cursor.getString(2),
-                    sponsor = cursor.getString(3),
-                    clickShim = cursor.getString(4),
-                    impressionShim = cursor.getString(5),
-                )
-            )
-        }
-    }
-}
diff --git a/components/service/pocket/src/test/resources/pocket/sponsored_stories_response.json b/components/service/pocket/src/test/resources/pocket/sponsored_stories_response.json
index a21f2fd7fef..45ca1e5b632 100644
--- a/components/service/pocket/src/test/resources/pocket/sponsored_stories_response.json
+++ b/components/service/pocket/src/test/resources/pocket/sponsored_stories_response.json
@@ -6,11 +6,13 @@
   "spocs": [
     {
       "id": 193815086,
+      "flight_id": 191739319,
       "campaign_id": 1315172,
       "title": "Eating Keto Has Never Been So Easy With Green Chef",
       "url": "https://i.geistm.com/l/GC_7ReasonsKetoV2_Journiest?bcid=601c567ac5b18a0414cce1d4&bhid=624f3ea9adad7604086ac6b3&utm_content=PKT_A_7ReasonsKetoV2_Journiest_40702022_RawMeatballUGC_130Off_601c567ac5b18a0414cce1d4_624f3ea9adad7604086ac6b3&tv=su4&ct=NAT-PK-PROS-130OFF5WEEK-037&utm_medium=DB&utm_source=pocket~geistm&utm_campaign=PKT_A_7ReasonsKetoV2_Journiest_40702022_RawMeatballUGC_130Off",
       "domain": "journiest.com",
       "excerpt": "Get Green Chef's Special Spring Offer: ${'$'}130 off plus free shipping.",
+      "priority": 3,
       "raw_image_src": "https://s.zkcdn.net/Advertisers/a3644de3c18948ffbd9aa43e8f9c7bf0.png",
       "image_src": "https://img-getpocket.cdn.mozilla.net/direct?url=realUrl.png&resize=w618-h310",
       "shim": {
@@ -19,15 +21,28 @@
         "delete": "193815086DeleteShim",
         "save": "193815086SaveShim"
       },
+      "caps": {
+        "lifetime": 50,
+        "campaign": {
+          "count": 10,
+          "period": 86400
+        },
+        "flight": {
+          "count": 10,
+          "period": 86400
+        }
+      },
       "sponsor": "Green Chef"
     },
     {
       "id": 177986195,
+      "flight_id": 191739667,
       "campaign_id": 63548984,
       "title": "This Leading Cash Back Card Is a Slam Dunk if You Want a One-Card Wallet",
       "url": "https://www.fool.com/the-ascent/credit-cards/landing/discover-it-cash-back-review-v2-csr/?utm_site=theascent&utm_campaign=ta-cc-co-pocket-discb-04012022-5-na-firefox&utm_medium=cpc&utm_source=pocket",
       "domain": "fool.com",
       "excerpt": "Make 2022 your year for a one-card wallet.",
+      "priority": 2,
       "raw_image_src": "https://s.zkcdn.net/Advertisers/359f56a5423c4926ab3aa148e448d839.webp",
       "image_src": "https://img-getpocket.cdn.mozilla.net/direct?url=https%3A//s.zkcdn.net/Advertisers/359f56a5423c4926ab3aa148e448d839.webp&resize=w618-h310",
       "shim": {
@@ -36,15 +51,28 @@
         "delete": "177986195DeleteShim",
         "save": "177986195SaveShim"
       },
+      "caps": {
+        "lifetime": 50,
+        "campaign": {
+          "count": 10,
+          "period": 86400
+        },
+        "flight": {
+          "count": 10,
+          "period": 86400
+        }
+      },
       "sponsor": "The Ascent"
     },
     {
       "id": 192560056,
+      "flight_id": 189212196,
       "campaign_id": 65544139,
       "title": "The Incredible Lawn Hack That Can Make Your Neighbors Green With Envy Over Your Lawn",
       "url": "https://go.lawnbuddy.org/zf/50/7673?campaign=SUN_Pocket2022&creative=SUN_LawnCompare4-TheIncredibleLawnHackThatCanMakeYourNeighborsGreenWithEnvyOverYourLawn-WithoutSpendingAFortuneOnNewGrassAndWithoutBreakingASweat-20220420",
       "domain": "go.lawnbuddy.org",
       "excerpt": "Without spending a fortune on new grass and without breaking a sweat.",
+      "priority": 1,
       "raw_image_src": "https://s.zkcdn.net/Advertisers/ce16302e184342cda0619c08b7604c9c.jpg",
       "image_src": "https://img-getpocket.cdn.mozilla.net/direct?url=https%3A//s.zkcdn.net/Advertisers/ce16302e184342cda0619c08b7604c9c.jpg&resize=w618-h310",
       "shim": {
@@ -53,6 +81,17 @@
         "delete": "192560056DeleteShim",
         "save": "192560056SaveShim"
       },
+      "caps": {
+        "lifetime": 50,
+        "campaign": {
+          "count": 10,
+          "period": 86400
+        },
+        "flight": {
+          "count": 10,
+          "period": 86400
+        }
+      },
       "sponsor": "Sunday"
     }
   ]
diff --git a/docs/changelog.md b/docs/changelog.md
index 4a18e3030a2..edc71ce3c8e 100644
--- a/docs/changelog.md
+++ b/docs/changelog.md
@@ -11,6 +11,10 @@ permalink: /changelog/
 * [Gecko](https://github.com/mozilla-mobile/android-components/blob/main/buildSrc/src/main/java/Gecko.kt)
 * [Configuration](https://github.com/mozilla-mobile/android-components/blob/main/.config.yml)
 
+* **service-pocket**
+  * 🌟️ Add support for rotating and pacing Pocket sponsored stories. [#12184](https://github.com/mozilla-mobile/android-components/issues/12184)
+  * See component's [README](https://github.com/mozilla-mobile/android-components/blob/main/components/service/pocket/README.md) to get more info.
+
 * **service-pocket**
   * Add a new `PocketStory` supertype for all Pocket stories. [#12171](https://github.com/mozilla-mobile/android-components/issues/12171)