func (w *DefaultWrapper) Unwrap(nw *NotificationWrap) (Notification, error) {
	var v Notification

	// Create struct depending on the type
	switch nw.Type {
	case "NewVulnerabilityNotification":
		v = &NewVulnerabilityNotification{}
	case "VulnerabilityPriorityIncreasedNotification":
		v = &VulnerabilityPriorityIncreasedNotification{}
	case "VulnerabilityPackageChangedNotification":
		v = &VulnerabilityPackageChangedNotification{}
	default:
		log.Warningf("could not unwrap notification [Type: %s]: unknown type for DefaultWrapper", nw.Type)
		return nil, cerrors.NewBadRequestError("could not unwrap notification")
	}

	// Unmarshal notification
	err := json.Unmarshal([]byte(nw.Data), v)
	if err != nil {
		log.Warningf("could not unmarshal notification with DefaultWrapper [Type: %s]: %s", nw.Type, err)
		return nil, cerrors.NewBadRequestError("could not unmarshal notification")
	}

	return v, nil
}

// Process detects the OS of a layer, the packages it installs/removes, and
// then stores everything in the database.
func Process(ID, parentID, path string) error {
	if ID == "" {
		return cerrors.NewBadRequestError("could not process a layer which does not have ID")
	}
	if path == "" {
		return cerrors.NewBadRequestError("could not process a layer which does not have a path")
	}

	log.Debugf("layer %s: processing (Location: %s, Engine version: %d, Parent: %s)", ID, utils.CleanURL(path), Version, parentID)

	// Check to see if the layer is already in the database.
	layer, err := database.FindOneLayerByID(ID, []string{database.FieldLayerEngineVersion})
	if err != nil && err != cerrors.ErrNotFound {
		return err
	}

	var parent *database.Layer

	if layer != nil {
		// The layer is already in the database, check if we need to update it.
		if layer.EngineVersion >= Version {
			log.Debugf("layer %s: layer content has already been processed in the past with engine %d. Current engine is %d. skipping analysis", ID, layer.EngineVersion, Version)
			return nil
		}

		log.Debugf("layer %s: layer content has been analyzed in the past with engine %d. Current engine is %d. analyzing again", ID, layer.EngineVersion, Version)
	} else {
		// The layer is a new one, create a base struct that we will fill.
		layer = &database.Layer{ID: ID, EngineVersion: Version}

		// Check to make sure that the parent's layer has already been processed.
		if parentID != "" {
			parent, err = database.FindOneLayerByID(parentID, []string{database.FieldLayerOS, database.FieldLayerPackages, database.FieldLayerPackages})
			if err != nil && err != cerrors.ErrNotFound {
				return err
			}
			if parent == nil {
				log.Warningf("layer %s: the parent layer (%s) is unknown. it must be processed first", ID, parentID)
				return ErrParentUnknown
			}
			layer.ParentNode = parent.GetNode()
		}
	}

	// Analyze the content.
	layer.OS, layer.InstalledPackagesNodes, layer.RemovedPackagesNodes, err = detectContent(ID, path, parent)
	if err != nil {
		return err
	}

	return database.InsertLayer(layer)
}

// GETLayersVulnerabilities returns the complete list of vulnerabilities that
// a layer has if it exists.
func GETLayersVulnerabilities(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
	// Get minumum priority parameter.
	minimumPriority := types.Priority(r.URL.Query().Get("minimumPriority"))
	if minimumPriority == "" {
		minimumPriority = "High" // Set default priority to High
	} else if !minimumPriority.IsValid() {
		httputils.WriteHTTPError(w, 0, cerrors.NewBadRequestError("invalid priority"))
		return
	}

	// Find layer
	layer, err := database.FindOneLayerByID(p.ByName("id"), []string{database.FieldLayerParent, database.FieldLayerPackages})
	if err != nil {
		httputils.WriteHTTPError(w, 0, err)
		return
	}

	// Find layer's packages.
	packagesNodes, err := layer.AllPackages()
	if err != nil {
		httputils.WriteHTTPError(w, 0, err)
		return
	}

	// Find vulnerabilities.
	vulnerabilities, err := getVulnerabilitiesFromLayerPackagesNodes(packagesNodes, minimumPriority, []string{database.FieldVulnerabilityID, database.FieldVulnerabilityLink, database.FieldVulnerabilityPriority, database.FieldVulnerabilityDescription, database.FieldVulnerabilityCausedByPackage})
	if err != nil {
		httputils.WriteHTTPError(w, 0, err)
		return
	}

	httputils.WriteHTTP(w, http.StatusOK, struct{ Vulnerabilities []*database.Vulnerability }{Vulnerabilities: vulnerabilities})
}

// DetectData finds the Data of the layer by using every registered DataDetector
func DetectData(path string, format string, toExtract []string, maxFileSize int64) (data map[string][]byte, err error) {
	var layerReader io.ReadCloser
	if strings.HasPrefix(path, "http://") || strings.HasPrefix(path, "https://") {
		r, err := http.Get(path)
		if err != nil {
			log.Warningf("could not download layer: %s", err)
			return nil, ErrCouldNotFindLayer
		}
		if math.Floor(float64(r.StatusCode/100)) != 2 {
			log.Warningf("could not download layer: got status code %d, expected 2XX", r.StatusCode)
			return nil, ErrCouldNotFindLayer
		}
		layerReader = r.Body
	} else {
		layerReader, err = os.Open(path)
		if err != nil {
			return nil, ErrCouldNotFindLayer
		}
	}
	defer layerReader.Close()

	for _, detector := range dataDetectors {
		if detector.Supported(path, format) {
			data, err = detector.Detect(layerReader, toExtract, maxFileSize)
			if err != nil {
				return nil, err
			}
			return data, nil
		}
	}

	return nil, cerrors.NewBadRequestError(fmt.Sprintf("unsupported image format '%s'", format))
}

// UpdateFlag creates a flag or update an existing flag's value
func UpdateFlag(name, value string) error {
	if name == "" || value == "" {
		log.Warning("could not insert a flag which has an empty name or value")
		return cerrors.NewBadRequestError("could not insert a flag which has an empty name or value")
	}

	// Initialize transaction
	t := cayley.NewTransaction()

	// Get current flag value
	currentValue, err := GetFlagValue(name)
	if err != nil {
		return err
	}

	// Build transaction
	name = "flag:" + name
	if currentValue != "" {
		t.RemoveQuad(cayley.Quad(name, "value", currentValue, ""))
	}
	t.AddQuad(cayley.Quad(name, "value", value, ""))

	// Apply transaction
	if err = store.ApplyTransaction(t); err != nil {
		log.Errorf("failed transaction (UpdateFlag): %s", err)
		return ErrTransaction
	}

	// Return
	return nil
}

func (pgSQL *pgSQL) insertNamespace(namespace database.Namespace) (int, error) {
	if namespace.Name == "" {
		return 0, cerrors.NewBadRequestError("could not find/insert invalid Namespace")
	}

	if pgSQL.cache != nil {
		promCacheQueriesTotal.WithLabelValues("namespace").Inc()
		if id, found := pgSQL.cache.Get("namespace:" + namespace.Name); found {
			promCacheHitsTotal.WithLabelValues("namespace").Inc()
			return id.(int), nil
		}
	}

	// We do `defer observeQueryTime` here because we don't want to observe cached namespaces.
	defer observeQueryTime("insertNamespace", "all", time.Now())

	var id int
	err := pgSQL.QueryRow(soiNamespace, namespace.Name).Scan(&id)
	if err != nil {
		return 0, handleError("soiNamespace", err)
	}

	if pgSQL.cache != nil {
		pgSQL.cache.Add("namespace:"+namespace.Name, id)
	}

	return id, nil
}

func parseConnectionString(source string) (dbName string, pgSourceURL string, err error) {
	if source == "" {
		return "", "", cerrors.NewBadRequestError("pgsql: no database connection string specified")
	}

	sourceURL, err := url.Parse(source)
	if err != nil {
		return "", "", cerrors.NewBadRequestError("pgsql: database connection string is not a valid URL")
	}

	dbName = strings.TrimPrefix(sourceURL.Path, "/")

	pgSource := *sourceURL
	pgSource.Path = "/postgres"
	pgSourceURL = pgSource.String()

	return
}

// InsertLayer insert a single layer in the database
//
// ID, and EngineVersion fields are required.
// ParentNode, OS, InstalledPackagesNodes and RemovedPackagesNodes are optional,
// SuccessorsNodes is unnecessary.
//
// The ID MUST be unique for two different layers.
//
//
// If the Layer already exists, nothing is done, except if the provided engine
// version is higher than the existing one, in which case, the OS,
// InstalledPackagesNodes and RemovedPackagesNodes fields will be replaced.
//
// The layer should only contains the newly installed/removed packages
// There is no safeguard that prevents from marking a package as newly installed
// while it has already been installed in one of its parent.
func InsertLayer(layer *Layer) error {
	// Verify parameters
	if layer.ID == "" {
		log.Warning("could not insert a layer which has an empty ID")
		return cerrors.NewBadRequestError("could not insert a layer which has an empty ID")
	}

	// Create required data structures
	t := cayley.NewTransaction()
	layer.Node = layer.GetNode()

	// Try to find an existing layer
	existingLayer, err := FindOneLayerByNode(layer.Node, FieldLayerAll)
	if err != nil && err != cerrors.ErrNotFound {
		return err
	}

	if existingLayer != nil && existingLayer.EngineVersion >= layer.EngineVersion {
		// The layer exists and has an equal or higher engine verison, do nothing
		return nil
	}

	if existingLayer == nil {
		// Create case: add permanent nodes
		t.AddQuad(cayley.Triple(layer.Node, fieldIs, fieldLayerIsValue))
		t.AddQuad(cayley.Triple(layer.Node, FieldLayerID, layer.ID))
		t.AddQuad(cayley.Triple(layer.Node, FieldLayerParent, layer.ParentNode))
	} else {
		// Update case: remove everything before we add updated data
		t.RemoveQuad(cayley.Triple(layer.Node, FieldLayerOS, existingLayer.OS))
		for _, pkg := range existingLayer.InstalledPackagesNodes {
			t.RemoveQuad(cayley.Triple(layer.Node, fieldLayerInstalledPackages, pkg))
		}
		for _, pkg := range existingLayer.RemovedPackagesNodes {
			t.RemoveQuad(cayley.Triple(layer.Node, fieldLayerRemovedPackages, pkg))
		}
		t.RemoveQuad(cayley.Triple(layer.Node, FieldLayerEngineVersion, strconv.Itoa(existingLayer.EngineVersion)))
	}

	// Add OS/Packages
	t.AddQuad(cayley.Triple(layer.Node, FieldLayerOS, layer.OS))
	for _, pkg := range layer.InstalledPackagesNodes {
		t.AddQuad(cayley.Triple(layer.Node, fieldLayerInstalledPackages, pkg))
	}
	for _, pkg := range layer.RemovedPackagesNodes {
		t.AddQuad(cayley.Triple(layer.Node, fieldLayerRemovedPackages, pkg))
	}
	t.AddQuad(cayley.Triple(layer.Node, FieldLayerEngineVersion, strconv.Itoa(layer.EngineVersion)))

	// Apply transaction
	if err = store.ApplyTransaction(t); err != nil {
		log.Errorf("failed transaction (InsertLayer): %s", err)
		return ErrTransaction
	}

	return nil
}

func (w *DefaultWrapper) Wrap(n Notification) (*NotificationWrap, error) {
	data, err := json.Marshal(n)
	if err != nil {
		log.Warningf("could not marshal notification [ID: %s, Type: %s]: %s", n.GetName(), n.GetType(), err)
		return nil, cerrors.NewBadRequestError("could not marshal notification with DefaultWrapper")
	}

	return &NotificationWrap{Type: n.GetType(), Data: string(data)}, nil
}

// NewHTTPNotifier initializes a new HTTPNotifier
func NewHTTPNotifier(URL string) (*HTTPNotifier, error) {
	if _, err := url.Parse(URL); err != nil {
		return nil, cerrors.NewBadRequestError("could not create a notifier with an invalid URL")
	}

	notifier := &HTTPNotifier{url: URL}
	health.RegisterHealthchecker("notifier", notifier.Healthcheck)

	return notifier, nil
}

// POSTBatchLayersVulnerabilities returns the complete list of vulnerabilities
// that the provided layers have, if they all exist.
func POSTBatchLayersVulnerabilities(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
	// Parse body
	var parameters POSTBatchLayersVulnerabilitiesParameters
	if s, err := httputils.ParseHTTPBody(r, &parameters); err != nil {
		httputils.WriteHTTPError(w, s, err)
		return
	}
	if len(parameters.LayersIDs) == 0 {
		httputils.WriteHTTPError(w, http.StatusBadRequest, errors.New("at least one LayerID query parameter must be provided"))
		return
	}

	// Get minumum priority parameter.
	minimumPriority := types.Priority(r.URL.Query().Get("minimumPriority"))
	if minimumPriority == "" {
		minimumPriority = "High" // Set default priority to High
	} else if !minimumPriority.IsValid() {
		httputils.WriteHTTPError(w, 0, cerrors.NewBadRequestError("invalid priority"))
		return
	}

	response := make(map[string]interface{})
	// For each LayerID parameter
	for _, layerID := range parameters.LayersIDs {
		// Find layer
		layer, err := database.FindOneLayerByID(layerID, []string{database.FieldLayerParent, database.FieldLayerPackages})
		if err != nil {
			httputils.WriteHTTPError(w, 0, err)
			return
		}

		// Find layer's packages.
		packagesNodes, err := layer.AllPackages()
		if err != nil {
			httputils.WriteHTTPError(w, 0, err)
			return
		}

		// Find vulnerabilities.
		vulnerabilities, err := getVulnerabilitiesFromLayerPackagesNodes(packagesNodes, minimumPriority, []string{database.FieldVulnerabilityID, database.FieldVulnerabilityLink, database.FieldVulnerabilityPriority, database.FieldVulnerabilityDescription, database.FieldVulnerabilityCausedByPackage})
		if err != nil {
			httputils.WriteHTTPError(w, 0, err)
			return
		}

		response[layerID] = struct{ Vulnerabilities []*database.Vulnerability }{Vulnerabilities: vulnerabilities}
	}

	httputils.WriteHTTP(w, http.StatusOK, response)
}

// DetectData finds the Data of the layer by using every registered DataDetector
func DetectData(format, path string, headers map[string]string, toExtract []string, maxFileSize int64) (data map[string][]byte, err error) {
	var layerReader io.ReadCloser
	if strings.HasPrefix(path, "http://") || strings.HasPrefix(path, "https://") {
		// Create a new HTTP request object.
		request, err := http.NewRequest("GET", path, nil)
		if err != nil {
			return nil, ErrCouldNotFindLayer
		}

		// Set any provided HTTP Headers.
		if headers != nil {
			for k, v := range headers {
				request.Header.Set(k, v)
			}
		}

		// Send the request and handle the response.
		r, err := http.DefaultClient.Do(request)
		if err != nil {
			log.Warningf("could not download layer: %s", err)
			return nil, ErrCouldNotFindLayer
		}

		// Fail if we don't receive a 2xx HTTP status code.
		if math.Floor(float64(r.StatusCode/100)) != 2 {
			log.Warningf("could not download layer: got status code %d, expected 2XX", r.StatusCode)
			return nil, ErrCouldNotFindLayer
		}

		layerReader = r.Body
	} else {
		layerReader, err = os.Open(path)
		if err != nil {
			return nil, ErrCouldNotFindLayer
		}
	}
	defer layerReader.Close()

	for _, detector := range dataDetectors {
		if detector.Supported(path, format) {
			data, err = detector.Detect(layerReader, toExtract, maxFileSize)
			if err != nil {
				return nil, err
			}
			return data, nil
		}
	}

	return nil, cerrors.NewBadRequestError(fmt.Sprintf("unsupported image format '%s'", format))
}

// GETLayersVulnerabilitiesDiff returns the list of vulnerabilities that a layer
// adds and removes if it exists.
func GETLayersVulnerabilitiesDiff(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
	// Get minumum priority parameter.
	minimumPriority := types.Priority(r.URL.Query().Get("minimumPriority"))
	if minimumPriority == "" {
		minimumPriority = "High" // Set default priority to High
	} else if !minimumPriority.IsValid() {
		httputils.WriteHTTPError(w, 0, cerrors.NewBadRequestError("invalid priority"))
		return
	}

	// Find layer.
	layer, err := database.FindOneLayerByID(p.ByName("id"), []string{database.FieldLayerPackages})
	if err != nil {
		httputils.WriteHTTPError(w, 0, err)
		return
	}

	// Selected fields for vulnerabilities.
	selectedFields := []string{database.FieldVulnerabilityID, database.FieldVulnerabilityLink, database.FieldVulnerabilityPriority, database.FieldVulnerabilityDescription, database.FieldVulnerabilityCausedByPackage}

	// Find vulnerabilities for installed packages.
	addedVulnerabilities, err := getVulnerabilitiesFromLayerPackagesNodes(layer.InstalledPackagesNodes, minimumPriority, selectedFields)
	if err != nil {
		httputils.WriteHTTPError(w, 0, err)
		return
	}

	// Find vulnerabilities for removed packages.
	removedVulnerabilities, err := getVulnerabilitiesFromLayerPackagesNodes(layer.RemovedPackagesNodes, minimumPriority, selectedFields)
	if err != nil {
		httputils.WriteHTTPError(w, 0, err)
		return
	}

	// Remove vulnerabilities which appears both in added and removed lists (eg. case of updated packages but still vulnerable).
	for ia, a := range addedVulnerabilities {
		for ir, r := range removedVulnerabilities {
			if a.ID == r.ID {
				addedVulnerabilities = append(addedVulnerabilities[:ia], addedVulnerabilities[ia+1:]...)
				removedVulnerabilities = append(removedVulnerabilities[:ir], removedVulnerabilities[ir+1:]...)
			}
		}
	}

	httputils.WriteHTTP(w, http.StatusOK, struct{ Adds, Removes []*database.Vulnerability }{Adds: addedVulnerabilities, Removes: removedVulnerabilities})
}

// FindLock returns the owner of a lock specified by its name and its
// expiration time.
func (pgSQL *pgSQL) FindLock(name string) (string, time.Time, error) {
	if name == "" {
		log.Warning("could not find an invalid lock")
		return "", time.Time{}, cerrors.NewBadRequestError("could not find an invalid lock")
	}

	defer observeQueryTime("FindLock", "all", time.Now())

	var owner string
	var until time.Time
	err := pgSQL.QueryRow(searchLock, name).Scan(&owner, &until)
	if err != nil {
		return owner, until, handleError("searchLock", err)
	}

	return owner, until, nil
}

// POSTVulnerabilities manually inserts a vulnerability into the database if it
// does not exist yet.
func POSTVulnerabilities(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
	var parameters *database.AbstractVulnerability
	if s, err := jsonhttp.ParseBody(r, &parameters); err != nil {
		jsonhttp.RenderError(w, s, err)
		return
	}

	// Ensure that the vulnerability does not exist.
	vulnerability, err := database.FindOneVulnerability(parameters.ID, []string{})
	if err != nil && err != cerrors.ErrNotFound {
		jsonhttp.RenderError(w, 0, err)
		return
	}
	if vulnerability != nil {
		jsonhttp.RenderError(w, 0, cerrors.NewBadRequestError("vulnerability already exists"))
		return
	}

	// Insert packages.
	packages := database.AbstractPackagesToPackages(parameters.AffectedPackages)
	err = database.InsertPackages(packages)
	if err != nil {
		jsonhttp.RenderError(w, 0, err)
		return
	}
	var pkgNodes []string
	for _, p := range packages {
		pkgNodes = append(pkgNodes, p.Node)
	}

	// Insert vulnerability.
	notifications, err := database.InsertVulnerabilities([]*database.Vulnerability{parameters.ToVulnerability(pkgNodes)})
	if err != nil {
		jsonhttp.RenderError(w, 0, err)
		return
	}

	// Insert notifications.
	err = database.InsertNotifications(notifications, database.GetDefaultNotificationWrapper())
	if err != nil {
		jsonhttp.RenderError(w, 0, err)
		return
	}

	jsonhttp.Render(w, http.StatusCreated, nil)
}

// InsertKeyValue stores (or updates) a single key / value tuple.
func (pgSQL *pgSQL) InsertKeyValue(key, value string) (err error) {
	if key == "" || value == "" {
		log.Warning("could not insert a flag which has an empty name or value")
		return cerrors.NewBadRequestError("could not insert a flag which has an empty name or value")
	}

	defer observeQueryTime("InsertKeyValue", "all", time.Now())

	// Upsert.
	//
	// Note: UPSERT works only on >= PostgreSQL 9.5 which is not yet supported by AWS RDS.
	//       The best solution is currently the use of http://dba.stackexchange.com/a/13477
	//       but the key/value storage doesn't need to be super-efficient and super-safe at the
	//       moment so we can just use a client-side solution with transactions, based on
	//       http://postgresql.org/docs/current/static/plpgsql-control-structures.html.
	// TODO(Quentin-M): Enable Upsert as soon as 9.5 is stable.

	for {
		// First, try to update.
		r, err := pgSQL.Exec(updateKeyValue, value, key)
		if err != nil {
			return handleError("updateKeyValue", err)
		}
		if n, _ := r.RowsAffected(); n > 0 {
			// Updated successfully.
			return nil
		}

		// Try to insert the key.
		// If someone else inserts the same key concurrently, we could get a unique-key violation error.
		_, err = pgSQL.Exec(insertKeyValue, key, value)
		if err != nil {
			if isErrUniqueViolation(err) {
				// Got unique constraint violation, retry.
				continue
			}
			return handleError("insertKeyValue", err)
		}

		return nil
	}
}

func (pgSQL *pgSQL) insertFeature(feature database.Feature) (int, error) {
	if feature.Name == "" {
		return 0, cerrors.NewBadRequestError("could not find/insert invalid Feature")
	}

	// Do cache lookup.
	if pgSQL.cache != nil {
		promCacheQueriesTotal.WithLabelValues("feature").Inc()
		id, found := pgSQL.cache.Get("feature:" + feature.Namespace.Name + ":" + feature.Name)
		if found {
			promCacheHitsTotal.WithLabelValues("feature").Inc()
			return id.(int), nil
		}
	}

	// We do `defer observeQueryTime` here because we don't want to observe cached features.
	defer observeQueryTime("insertFeature", "all", time.Now())

	// Find or create Namespace.
	namespaceID, err := pgSQL.insertNamespace(feature.Namespace)
	if err != nil {
		return 0, err
	}

	// Find or create Feature.
	var id int
	err = pgSQL.QueryRow(soiFeature, feature.Name, namespaceID).Scan(&id)
	if err != nil {
		return 0, handleError("soiFeature", err)
	}

	if pgSQL.cache != nil {
		pgSQL.cache.Add("feature:"+feature.Namespace.Name+":"+feature.Name, id)
	}

	return id, nil
}

func (pgSQL *pgSQL) insertFeatureVersion(featureVersion database.FeatureVersion) (id int, err error) {
	if featureVersion.Version.String() == "" {
		return 0, cerrors.NewBadRequestError("could not find/insert invalid FeatureVersion")
	}

	// Do cache lookup.
	cacheIndex := "featureversion:" + featureVersion.Feature.Namespace.Name + ":" + featureVersion.Feature.Name + ":" + featureVersion.Version.String()
	if pgSQL.cache != nil {
		promCacheQueriesTotal.WithLabelValues("featureversion").Inc()
		id, found := pgSQL.cache.Get(cacheIndex)
		if found {
			promCacheHitsTotal.WithLabelValues("featureversion").Inc()
			return id.(int), nil
		}
	}

	// We do `defer observeQueryTime` here because we don't want to observe cached featureversions.
	defer observeQueryTime("insertFeatureVersion", "all", time.Now())

	// Find or create Feature first.
	t := time.Now()
	featureID, err := pgSQL.insertFeature(featureVersion.Feature)
	observeQueryTime("insertFeatureVersion", "insertFeature", t)

	if err != nil {
		return 0, err
	}

	featureVersion.Feature.ID = featureID

	// Begin transaction.
	tx, err := pgSQL.Begin()
	if err != nil {
		tx.Rollback()
		return 0, handleError("insertFeatureVersion.Begin()", err)
	}

	// Lock Vulnerability_Affects_FeatureVersion exclusively.
	// We want to prevent InsertVulnerability to modify it.
	promConcurrentLockVAFV.Inc()
	defer promConcurrentLockVAFV.Dec()
	t = time.Now()
	_, err = tx.Exec(lockVulnerabilityAffects)
	observeQueryTime("insertFeatureVersion", "lock", t)

	if err != nil {
		tx.Rollback()
		return 0, handleError("insertFeatureVersion.lockVulnerabilityAffects", err)
	}

	// Find or create FeatureVersion.
	var newOrExisting string

	t = time.Now()
	err = tx.QueryRow(soiFeatureVersion, featureID, &featureVersion.Version).
		Scan(&newOrExisting, &featureVersion.ID)
	observeQueryTime("insertFeatureVersion", "soiFeatureVersion", t)

	if err != nil {
		tx.Rollback()
		return 0, handleError("soiFeatureVersion", err)
	}

	if newOrExisting == "exi" {
		// That featureVersion already exists, return its id.
		tx.Commit()

		if pgSQL.cache != nil {
			pgSQL.cache.Add(cacheIndex, featureVersion.ID)
		}

		return featureVersion.ID, nil
	}

	// Link the new FeatureVersion with every vulnerabilities that affect it, by inserting in
	// Vulnerability_Affects_FeatureVersion.
	t = time.Now()
	err = linkFeatureVersionToVulnerabilities(tx, featureVersion)
	observeQueryTime("insertFeatureVersion", "linkFeatureVersionToVulnerabilities", t)

	if err != nil {
		tx.Rollback()
		return 0, err
	}

	// Commit transaction.
	err = tx.Commit()
	if err != nil {
		return 0, handleError("insertFeatureVersion.Commit()", err)
	}

	if pgSQL.cache != nil {
		pgSQL.cache.Add(cacheIndex, featureVersion.ID)
	}

	return featureVersion.ID, nil
}

// InsertVulnerabilities inserts or updates several vulnerabilities in the database in one transaction
// During an update, if the vulnerability was previously fixed by a version in a branch and a new package of that branch is specified, the previous one is deleted
// Otherwise, it simply adds the defined packages, there is currently no way to delete affected packages.
//
// ID, Link, Priority and FixedInNodes fields have to be specified. Description is optionnal.
func InsertVulnerabilities(vulnerabilities []*Vulnerability) ([]Notification, error) {
	if len(vulnerabilities) == 0 {
		return []Notification{}, nil
	}

	// Create required data structure
	var err error
	t := cayley.NewTransaction()
	cachedVulnerabilities := make(map[string]*Vulnerability)

	var notifications []Notification
	newVulnerabilityNotifications := make(map[string]*NewVulnerabilityNotification)
	vulnerabilityPriorityIncreasedNotifications := make(map[string]*VulnerabilityPriorityIncreasedNotification)
	vulnerabilityPackageChangedNotifications := make(map[string]*VulnerabilityPackageChangedNotification)

	// Iterate over all the vulnerabilities we need to insert/update
	for _, vulnerability := range vulnerabilities {
		// Check if the vulnerability already exists
		existingVulnerability, _ := cachedVulnerabilities[vulnerability.ID]
		if existingVulnerability == nil {
			existingVulnerability, err = FindOneVulnerability(vulnerability.ID, FieldVulnerabilityAll)
			if err != nil && err != cerrors.ErrNotFound {
				return []Notification{}, err
			}
			if existingVulnerability != nil {
				cachedVulnerabilities[vulnerability.ID] = existingVulnerability
			}
		}

		// Insert/Update vulnerability
		if existingVulnerability == nil {
			// The vulnerability does not exist, create it

			// Verify parameters
			if vulnerability.ID == "" || vulnerability.Link == "" || vulnerability.Priority == "" {
				log.Warningf("could not insert an incomplete vulnerability [ID: %s, Link: %s, Priority: %s]", vulnerability.ID, vulnerability.Link, vulnerability.Priority)
				return []Notification{}, cerrors.NewBadRequestError("Could not insert an incomplete vulnerability")
			}
			if !vulnerability.Priority.IsValid() {
				log.Warningf("could not insert a vulnerability which has an invalid priority [ID: %s, Link: %s, Priority: %s]. Valid priorities are: %v.", vulnerability.ID, vulnerability.Link, vulnerability.Priority, types.Priorities)
				return []Notification{}, cerrors.NewBadRequestError("Could not insert a vulnerability which has an invalid priority")
			}
			if len(vulnerability.FixedInNodes) == 0 {
				log.Warningf("could not insert a vulnerability which doesn't affect any package [ID: %s].", vulnerability.ID)
				return []Notification{}, cerrors.NewBadRequestError("could not insert a vulnerability which doesn't affect any package")
			}

			// Insert it
			vulnerability.Node = vulnerability.GetNode()

			t.AddQuad(cayley.Triple(vulnerability.Node, fieldIs, fieldVulnerabilityIsValue))
			t.AddQuad(cayley.Triple(vulnerability.Node, FieldVulnerabilityID, vulnerability.ID))
			t.AddQuad(cayley.Triple(vulnerability.Node, FieldVulnerabilityLink, vulnerability.Link))
			t.AddQuad(cayley.Triple(vulnerability.Node, FieldVulnerabilityPriority, string(vulnerability.Priority)))
			t.AddQuad(cayley.Triple(vulnerability.Node, FieldVulnerabilityDescription, vulnerability.Description))
			for _, p := range vulnerability.FixedInNodes {
				t.AddQuad(cayley.Triple(vulnerability.Node, FieldVulnerabilityFixedIn, p))
			}

			// Add a notification
			notification := &NewVulnerabilityNotification{VulnerabilityID: vulnerability.ID}
			notifications = append(notifications, notification)
			newVulnerabilityNotifications[vulnerability.ID] = notification

			cachedVulnerabilities[vulnerability.ID] = vulnerability
		} else {
			// The vulnerability already exists, update it
			if vulnerability.Link != "" && existingVulnerability.Link != vulnerability.Link {
				t.RemoveQuad(cayley.Triple(existingVulnerability.Node, FieldVulnerabilityLink, existingVulnerability.Link))
				t.AddQuad(cayley.Triple(existingVulnerability.Node, FieldVulnerabilityLink, vulnerability.Link))
				existingVulnerability.Link = vulnerability.Link
			}

			if vulnerability.Priority != "" && vulnerability.Priority != types.Unknown && existingVulnerability.Priority != vulnerability.Priority {
				if !vulnerability.Priority.IsValid() {
					log.Warningf("could not update a vulnerability which has an invalid priority [ID: %s, Link: %s, Priority: %s]. Valid priorities are: %v.", vulnerability.ID, vulnerability.Link, vulnerability.Priority, types.Priorities)
					return []Notification{}, cerrors.NewBadRequestError("Could not update a vulnerability which has an invalid priority")
				}

				// Add a notification about the priority change if the new priority is higher and the vulnerability is not new
				if vulnerability.Priority.Compare(existingVulnerability.Priority) > 0 {
					if _, newVulnerabilityNotificationExists := newVulnerabilityNotifications[vulnerability.ID]; !newVulnerabilityNotificationExists {
						// Any priorityChangeNotification already ?
						if existingPriorityNotification, _ := vulnerabilityPriorityIncreasedNotifications[vulnerability.ID]; existingPriorityNotification != nil {
							// There is a priority change notification, replace it but keep the old priority field
							existingPriorityNotification.NewPriority = vulnerability.Priority
						} else {
							// No previous notification, just add a new one
							notification := &VulnerabilityPriorityIncreasedNotification{OldPriority: existingVulnerability.Priority, NewPriority: vulnerability.Priority, VulnerabilityID: existingVulnerability.ID}
							notifications = append(notifications, notification)
							vulnerabilityPriorityIncreasedNotifications[vulnerability.ID] = notification
						}
					}
				}

//.........这里部分代码省略.........

// Process detects the Namespace of a layer, the features it adds/removes, and
// then stores everything in the database.
// TODO(Quentin-M): We could have a goroutine that looks for layers that have been analyzed with an
// older engine version and that processes them.
func Process(datastore database.Datastore, imageFormat, name, parentName, path string, headers map[string]string) error {
	// Verify parameters.
	if name == "" {
		return cerrors.NewBadRequestError("could not process a layer which does not have a name")
	}

	if path == "" {
		return cerrors.NewBadRequestError("could not process a layer which does not have a path")
	}

	if imageFormat == "" {
		return cerrors.NewBadRequestError("could not process a layer which does not have a format")
	}

	log.Debugf("layer %s: processing (Location: %s, Engine version: %d, Parent: %s, Format: %s)",
		name, utils.CleanURL(path), Version, parentName, imageFormat)

	// Check to see if the layer is already in the database.
	layer, err := datastore.FindLayer(name, false, false)
	if err != nil && err != cerrors.ErrNotFound {
		return err
	}

	if err == cerrors.ErrNotFound {
		// New layer case.
		layer = database.Layer{Name: name, EngineVersion: Version}

		// Retrieve the parent if it has one.
		// We need to get it with its Features in order to diff them.
		if parentName != "" {
			parent, err := datastore.FindLayer(parentName, true, false)
			if err != nil && err != cerrors.ErrNotFound {
				return err
			}
			if err == cerrors.ErrNotFound {
				log.Warningf("layer %s: the parent layer (%s) is unknown. it must be processed first", name,
					parentName)
				return ErrParentUnknown
			}
			layer.Parent = &parent
		}
	} else {
		// The layer is already in the database, check if we need to update it.
		if layer.EngineVersion >= Version {
			log.Debugf(`layer %s: layer content has already been processed in the past with engine %d.
        Current engine is %d. skipping analysis`, name, layer.EngineVersion, Version)
			return nil
		}

		log.Debugf(`layer %s: layer content has been analyzed in the past with engine %d. Current
      engine is %d. analyzing again`, name, layer.EngineVersion, Version)
	}

	// Analyze the content.
	layer.Namespace, layer.Features, err = detectContent(imageFormat, name, path, headers, layer.Parent)
	if err != nil {
		return err
	}

	return datastore.InsertLayer(layer)
}